Utilities for working with python services.
Default configuration for structlog.
Configure json logging at the INFO level:
from servicetools.logging_config import default_logging, LogFormat, Verbosity
default_logging(Verbosity.INFO, LogFormat.JSON)
Configure text logging at the DEBUG level:
from servicetools.logging_config import default_logging, LogFormat, Verbosity
default_logging(Verbosity.DEBUG, LogFormat.TEXT)
Configure text logging at the DEBUG level and filter out external loggers:
from servicetools.logging_config import default_logging, LogFormat, Verbosity
default_logging(Verbosity.DEBUG, LogFormat.TEXT, ["extern_logger_1"])
Decorator to add timing information to the logs:
from servicetools.timer import timer
import structlog
@timer(structlog.get_logger(__name__))
def some_function():
pass
Create namespace relative patches:
import some_package.sub_package.another_package as under_test
from servicetools.testing import relative_patch_maker
patch = relative_patch_maker(under_test.__name__)
class TestStuff:
#equivalent to @unittest.mock.patch("some_package.sub_package.another_package.something_to_patch")
@patch("something_to_patch")
def test_something(self, patched):
under_test.something()
patched.assert_called_once()
#equivalent to @unittest.mock.patch("some_package.sub_package.another_package.something_else_to_patch")
@patch("something_else_to_patch")
def test_something(self, patched):
under_test.something()
patched.assert_called_once()
Middleware for Starlette framework to log HTTP requests to structlog. Log entries will be made at the start and end of each request. Error requests (400s and 500s) will also be logged. Any calls that throw exceptions will be converted 500 responses.
from servicetools.middleware import StructlogRequestMiddleware
import structlog
app.add_middleware(StructlogRequestMiddleware(app, logger=structlog.get_logger(__name__)))
There are options to customize the logging:
import logging
import structlog
from servicetools.middleware import StructlogRequestMiddleware
app.add_middleware(StructlogRequestMiddleware(
app,
logger=structlog.get_logger(__name__),
log_level=logging.DEBUG, # Log at the DEBUG level.
ignored_status_codes={404}, # Do not log 404 errors.
))
Specification for dramatiq actors that allows them to connect a broker
explicitly through the init_actor
function rather than implicitly when they are created. This allows
you to defer setting up your Rabbitmq broker to a time of your choosing. To create a new actor that
uses this behavior, set up your dramatiq actors like so
import dramatiq
from servicetools.lazyactor import LazyActor
@dramatiq.actor(actor_class=LazyActor)
def test_func(data: str) -> None:
print(data)
Now, whenever you have set up your Rabbitmq instance and connected to it, tell your actors to connect to it like so
from dramatiq.brokers.rabbitmq import RabbitmqBroker
from pika import PlainCredentials
import dramatiq
broker = RabbitmqBroker(
host="localhost",
credentials=PlainCredentials(username="user", password="password"),
)
dramatiq.set_broker(broker)
test_func.init_actor(broker=broker)
This project uses poetry:
$ pip install poetry
$ cd to/project/root
$ poetry install
Testing is done via pytest.
$ poetry run pytest