structlog

A non-destructive structlog processor that forwards events to axonpush as OpenTelemetry-shaped app.log events.

pip install "axonpush[structlog]"

Tested against structlog>=24.0,<26.

Add the processor

import structlog
from axonpush import AxonPush
from axonpush.integrations.structlog import axonpush_structlog_processor

client = AxonPush()

forwarder = axonpush_structlog_processor(
    client=client,
    channel_id=channel_id,
    service_name="my-api",
    environment="production",
)

structlog.configure(
    processors=[
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        forwarder,
        structlog.processors.JSONRenderer(),
    ],
)

The processor is non-destructive: it returns the event dict unchanged, so everything downstream of it sees exactly what it would have seen.

Place it before the renderer. JSONRenderer, KeyValueRenderer and ConsoleRenderer collapse the event dict into a string, and after that the structured fields are gone. Put forwarder after the processors that enrich the dict, add_log_level, TimeStamper, and before the one that renders it.

Then log normally:

log = structlog.get_logger("my_app")

log.info("user signed in", user_id=42, method="oauth")
log.error("downstream timeout", endpoint="/api/search", elapsed_ms=5000)

request_log = log.bind(request_id="req-9f21")
request_log.info("handling request", path="/chat")

Constructor

axonpush_structlog_processor(
    *,
    client,                  # AxonPush or AsyncAxonPush - required
    channel_id,              # str UUID - required
    source="app",            # "app" -> app.log, "agent" -> agent.log
    service_name=None,
    service_version=None,
    environment=None,
    agent_id=None,
    mode=None,               # "background" (default) | "sync"
    queue_size=1000,
    shutdown_timeout=2.0,
)

Every argument is keyword-only, and source and mode are validated at construction.

In "background" mode the processor pushes onto a bounded queue drained by one daemon thread, so the logging call stays O(microseconds). Pass a sync AxonPush, an AsyncAxonPush gets no worker and publishes inline.

What each event becomes

Field	Value
`identifier`	The literal `structlog`
`event_type`	`app.log`, or `agent.log` when `source="agent"`
`payload.body`	The `event` key from the event dict
`payload.severityNumber` / `payload.severityText`	OTel severity from the `level` key, falling back to the method name
`payload.timeUnixNano`	Parsed from `timestamp` or `time`, ISO strings and epoch floats both work; wall-clock now if neither is present
`payload.attributes`	Every other key in the event dict, including bound context
`payload.resource`	`service.name`, `service.version`, `deployment.environment` when configured

The identifier is a constant, not the logger name. structlog does not put the logger name in the event dict by default, so filter by payload.attributes.logger, add structlog.stdlib.add_logger_name to the chain if you want that key populated.

The trace id comes from the active trace context, so events emitted inside a traced request join that trace.

Flushing

forwarder.flush(timeout=1.0)   # block until the queue drains
forwarder.close()              # drain and stop the worker

For Lambda and other freeze-between-invocations runtimes, flush_after_invocation is re-exported from this module and behaves as it does for the stdlib handler:

from axonpush.integrations.structlog import (
    axonpush_structlog_processor,
    flush_after_invocation,
)

@flush_after_invocation(forwarder)
def lambda_handler(event, context):
    ...

Add the processor

Constructor

What each event becomes

Flushing

On this page