Frameworks#

To have consistent log output, it makes sense to configure structlog before any logging is done. The best place to perform your configuration varies with applications and frameworks. If you use standard library’s logging, it makes sense to configure them next to each other.

OpenTelemetry#

The Python OpenTelemetry SDK offers an easy API to get the current span, so you can enrich your logs with a straight-forward processor:

from opentelemetry import trace

def add_open_telemetry_spans(_, __, event_dict):
    span = trace.get_current_span()
    if not span.is_recording():
        event_dict["span"] = None
        return event_dict

    ctx = span.get_span_context()
    parent = getattr(span, "parent", None)

    event_dict["span"] = {
        "span_id": hex(ctx.span_id),
        "trace_id": hex(ctx.trace_id),
        "parent_span_id": None if not parent else hex(parent.span_id),
    }

    return event_dict

Django#

django-structlog is a popular and well-maintained package that does all the heavy lifting.

Flask#

See Flask’s Logging docs.

Generally speaking: configure structlog before instantiating flask.Flask.

Here’s a signal handler that binds various request details into context variables:

def bind_request_details(sender: Flask, **extras: dict[str, Any]) -> None:
    structlog.contextvars.clear_contextvars()
    structlog.contextvars.bind_contextvars(
        request_id=request.headers.get("X-Unique-ID", "NONE"),
        peer=peer,
    )

    if current_user.is_authenticated:
        structlog.contextvars.bind_contextvars(
            user_id=current_user.get_id(),
        )

You add it to an existing app like this:

from flask import request_started

request_started.connect(bind_request_details, app)

Pyramid#

Configure it in the application constructor.

Here’s an example for a Pyramid Tween that stores various request-specific data into context variables:

@dataclass
class StructLogTween:
    handler: Callable[[Request], Response]
    registry: Registry

    def __call__(self, request: Request) -> Response:
        structlog.contextvars.clear_contextvars()
        structlog.contextvars.bind_contextvars(
            peer=request.client_addr,
            request_id=request.headers.get("X-Unique-ID", "NONE"),
            user_agent=request.environ.get("HTTP_USER_AGENT", "UNKNOWN"),
            user=request.authenticated_userid,
        )

        return self.handler(request)

Celery#

Celery’s multi-process architecture leads unavoidably to race conditions that show up as interleaved logs. It ships standard library-based helpers in the form of celery.utils.log.get_task_logger() that you should use inside of tasks to prevent that problem.

The most straight-forward way to integrate that with structlog is using Standard Library Logging and wrapping that logger using structlog.wrap_logger():

from celery.utils.log import get_task_logger

logger = structlog.wrap_logger(get_task_logger(__name__))

If you want to automatically bind task metadata to your Context Variables, you can use Celery’s signals:

from celery import signals

@signals.task_prerun.connect
def on_task_prerun(sender, task_id, task, args, kwargs, **_):
    structlog.contextvars.bind_contextvars(task_id=task_id, task_name=task.name)

See this issue for more details.

Twisted#

The plugin definition is the best place. If your app is not a plugin, put it into your tac file.