Migrating from Standard Logging to structlog

You need to move a running Python service from standard library logging calls to structured JSON output from structlog without double-emitting records, losing context, or taking a maintenance window for the cutover. This page is for backend engineers and SREs who already have a service in production, a working handler topology, and thousands of existing call sites they would rather not rewrite. It is a focused task within the structlog architecture and setup reference, part of the Modern Python Logging Libraries Deep Dive guide, and it assumes you have already decided structlog is the right destination — if you have not, the trade-offs are laid out in structlog vs Loguru vs standard library logging.

Prerequisites

Pin structlog and an OpenTelemetry API for the trace-context step. Nothing else is required: the context store the migration relies on is the standard library's contextvars module.

pip install \
  "structlog>=24.1.0,<26.0.0" \
  "opentelemetry-api>=1.30.0,<2.0.0"

Run configuration at import time, before any logger is fetched, so cached loggers pick up the chain. Set the service log level through the environment if you gate output by severity; the mapping between those names and the numeric thresholds standard logging compares against is covered in log levels and severity mapping.

export LOG_LEVEL="INFO"

Implementation

The migration is zero-downtime precisely because the stdlib factory bridges rather than replaces: existing logging.getLogger(__name__).info(...) calls keep working and start emitting JSON the moment configuration runs, so you can cut over a running service by deploying a configuration change rather than rewriting call sites. The risk is not the call sites; it is the handler topology. Standard logging propagates records up to the root logger, and any library that logs — your web framework, the database driver, an HTTP client — feeds the same root. If both a leftover root handler and the structlog pipeline are attached, every one of those records emits twice. Reviewing the existing handler architecture before you start is worth the ten minutes: you need to know which handlers exist, which loggers set propagate = False, and where the records currently land. The first step then removes the existing handlers before anything else.

Step 1 — Clear root handlers and install the stdlib factory. Standard logging attaches a default StreamHandler to the root logger. Adding a structlog handler on top double-emits, so remove the existing handlers first, then configure structlog with structlog.stdlib.LoggerFactory and BoundLogger. This routes legacy logging.getLogger calls through the same processor chain without touching call sites.

import logging
import structlog

# Remove the default root handler to prevent double emission
for handler in logging.root.handlers[:]:
    logging.root.removeHandler(handler)

structlog.configure(
    wrapper_class=structlog.stdlib.BoundLogger,
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

structlog.get_logger().info("service_started", version="2.1.0")

Expected Output:

{"event": "service_started", "level": "info", "logger": "__main__", "timestamp": "2024-01-15T10:00:00.000000Z", "version": "2.1.0"}

Step 2 — Order the processor chain and inject trace context. Processor order dictates serialization: merge async context first, add level and timestamp, format exceptions and stack info before the renderer, then render JSON last. Inject trace_id and span_id via bind_contextvars so they survive await boundaries.

import structlog
from opentelemetry import trace

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer(),
    ],
)


def inject_otel_context():
    ctx = trace.get_current_span().get_span_context()
    if ctx.is_valid:
        structlog.contextvars.bind_contextvars(
            trace_id=f"{ctx.trace_id:032x}",
            span_id=f"{ctx.span_id:016x}",
            trace_flags=int(ctx.trace_flags),
        )


inject_otel_context()
structlog.get_logger().info("request_processed", endpoint="/api/v1/data")

Expected Output:

{"event": "request_processed", "level": "info", "timestamp": "2024-01-15T10:00:01.000000Z", "trace_id": "00000000000000000000000000000001", "span_id": "0000000000000002", "trace_flags": 1, "endpoint": "/api/v1/data"}

Step 3 — Handle legacy formatting and exception traces. Legacy code passes positional %s arguments; structlog expects keyword arguments for clean JSON keys, and BoundLogger auto-converts the positional string. Multi-line tracebacks need structlog.processors.format_exc_info in the chain; exc_info=True alone does not serialize frames into the exception key. Validate the schema so a stack_info key does not collide with exception.

For a phased rollout across many services, run the new pipeline behind a feature flag and direct the structlog JSON stream to stdout while leaving any legacy file handler in place but isolated, so the two never share a destination. Once you have confirmed the JSON parses cleanly in the aggregator and the trace fields correlate to spans, remove the legacy handler. The contextvars-based trace injection in step two is what makes the migrated logs useful rather than merely structured: a record without trace_id is structured noise, while a record carrying the active span's identifiers joins directly to the trace waterfall in your backend — the same correlation problem solved from the standard library side in adding trace IDs to log records, and end-to-end across services by context propagation and baggage. Bind those identifiers once at request entry — in middleware or an ASGI lifespan dependency — and merge_contextvars carries them into every record the request produces, including records emitted by bridged stdlib loggers deep in third-party code. The binding, clearing and nesting rules for that store are covered in detail in binding context variables in structlog.

Incremental migration with ProcessorFormatter

The configuration in step one routes records through structlog's own LoggerFactory, which is the right end state but a large blast radius for a first deploy. The lower-risk path keeps the standard library's handler topology exactly as it is and changes only the formatter. A structlog.stdlib.ProcessorFormatter can render any LogRecord — including ones that never touched structlog — through a shared processor chain, so you flip the output to JSON before you change a single call site, then adopt structlog.get_logger module by module on your own schedule.

The key is one shared processor chain used in two places: as the foreign_pre_chain for foreign stdlib records, and as the prefix of the native structlog chain. Define it once so the two paths can never drift.

import logging
import structlog

# One chain, reused for both native structlog events and foreign stdlib records.
shared_processors = [
    structlog.contextvars.merge_contextvars,
    structlog.stdlib.add_logger_name,
    structlog.stdlib.add_log_level,
    structlog.processors.TimeStamper(fmt="iso"),
    structlog.processors.StackInfoRenderer(),
    structlog.processors.format_exc_info,
]

# Native structlog ends by handing the event to the stdlib formatter.
structlog.configure(
    processors=shared_processors + [
        structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
    ],
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

# One ProcessorFormatter renders BOTH sources to flat JSON.
formatter = structlog.stdlib.ProcessorFormatter(
    foreign_pre_chain=shared_processors,   # replays the chain on non-structlog records
    processors=[
        structlog.stdlib.ProcessorFormatter.remove_processors_meta,
        structlog.processors.JSONRenderer(),
    ],
)

handler = logging.StreamHandler()
handler.setFormatter(formatter)
root = logging.getLogger()
root.handlers[:] = [handler]   # replace, do not append, to avoid double emission
root.setLevel(logging.INFO)

# Foreign caller that has NOT been migrated yet — still emits JSON.
logging.getLogger("legacy.payments").warning("charge_retry", extra={"attempt": 2})
# Already-migrated caller.
structlog.get_logger("orders").info("order_created", order_id="ORD-9")

Expected Output:

{"attempt": 2, "logger": "legacy.payments", "level": "warning", "timestamp": "2024-01-15T10:00:02.000000Z", "event": "charge_retry"}
{"logger": "orders", "level": "info", "timestamp": "2024-01-15T10:00:02.001000Z", "order_id": "ORD-9", "event": "order_created"}

Both lines share the same keys, the same timestamp format, and the same level vocabulary because they ran the same shared_processors. That is what keeps third-party libraries consistent: a SQL warning from your ORM, a connection-pool message from an HTTP client, and your own migrated events all reach the aggregator as one schema, so a single parser and a single set of dashboard fields cover everything. You never have to migrate the library; its LogRecord simply flows through the foreign_pre_chain. The rollout then becomes safe and monotonic — ship the formatter swap first and watch the JSON validate, then convert call sites whenever a module is already being touched, with no flag-day cutover and no window where half the stream is plain text and half is JSON.

If your service already declares its handlers through dictConfig, you do not have to abandon that declaration to adopt this pattern. Keep the dictionary and replace only the formatters entry, pointing () at a factory that returns the ProcessorFormatter above; handler classes, levels, per-logger propagate flags and filters stay exactly where they were, which means the diff a reviewer reads is a handful of lines rather than a rewritten bootstrap module. The same applies if you route through a QueueHandler for non-blocking logging: put the ProcessorFormatter on the handler owned by the listener, not on the queue-side handler, so serialization happens on the listener thread and the request path only pays for enqueueing. For the Django-specific shape of that same dictionary, see structlog JSON logging in Django.

Configuration options

The only entry with a genuine trade-off is cache_logger_on_first_use. Caching binds the resolved processor chain to the logger on first call, which is what keeps per-record cost flat under load, but it also means a later structlog.configure() will not reach loggers that have already been used. Leave it True in production and False in tests, where reconfiguring between cases is the whole point. When you are weighing this migration against a trace-correlated microservice deployment specifically, Loguru vs structlog for microservices covers how the two libraries differ once per-request context becomes a hard requirement.

Verification

Use structlog.testing.LogCapture in CI to assert the structured record without serializing to a real sink. This catches schema regressions before deploy.

import structlog.testing

capture = structlog.testing.LogCapture()
structlog.configure(
    processors=[capture],
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=False,
)

structlog.get_logger().info("test_migration", status="pass")

assert len(capture.entries) == 1
assert capture.entries[0]["status"] == "pass"
print("Validation successful.")

Expected Output:

Validation successful.

LogCapture proves the native path. The migration-specific regression it cannot catch is a foreign record drifting from a native one, so add a second assertion that goes through the real handler. Capture stdout with pytest's capsys (or a logging.StreamHandler pointed at an io.StringIO), emit one logging.getLogger(...) line and one structlog.get_logger(...) line, parse both with json.loads, and assert their key sets match. That single test is what stops a half-migrated service from shipping two schemas to the aggregator: it fails the moment someone adds a processor to structlog.configure but forgets the foreign_pre_chain. Run it against the same bootstrap function production calls, not a chain redefined in the test file, or you are only testing the test.

Before the cutover deploy, do one manual check as well: run the service, pipe stdout through a JSON parser, and confirm that no line fails to parse and that every line carries timestamp, level, logger and — inside a request — trace_id. A single unparsable line usually means a rogue print() or a library writing straight to stderr, and it is far cheaper to find it locally than in the ingest pipeline's dead-letter queue.

Common mistakes

• Error signature: {"message": "{\"event\": \"test\", \"level\": \"info\"}"} — valid JSON whose payload is an escaped JSON string. Root cause: both a logging.Formatter and the structlog JSONRenderer serialize the same record, so it is encoded twice. Remediation: remove all formatters from the handlers and let structlog own serialization, either through the JSONRenderer at the end of the chain or through a single ProcessorFormatter.

• Error signature: TypeError: 'Logger' object is not callable, or output missing processors that are plainly listed in the configuration. Root cause: structlog.configure() ran after the first get_logger(), and with cache_logger_on_first_use=True those cached loggers still hold the old chain. Remediation: configure at import time in a bootstrap module imported before anything that logs, and never reconfigure at runtime.

• Error signature: exception records arrive with a message but no exception key, most often from async def handlers. Root cause: format_exc_info is absent from the chain, or the log call runs outside the except block where sys.exc_info() is still populated. Remediation: add format_exc_info before the renderer and call the logger with exc_info=True from inside the exception handler.

• Error signature: every record appears twice in the aggregator, once as JSON and once as plain text. Root cause: a handler was appended to the root logger instead of replacing the existing ones, or a child logger keeps propagate = True while also owning its own handler. Remediation: assign root.handlers[:] = [handler] rather than calling addHandler, and audit any logger that sets both a handler and propagation.

• Error signature: third-party LogRecord lines lack timestamp or carry a different level field than your migrated calls. Root cause: the foreign_pre_chain and the native structlog processors were defined as two separate lists that fell out of sync. Remediation: define one shared_processors list and reference it in both structlog.configure and ProcessorFormatter(foreign_pre_chain=...), so the two code paths cannot diverge.

Related

• structlog architecture and setup — the parent reference on the processor pipeline, renderer selection and bootstrap order this migration lands on.
• Binding context variables in structlog — the next step once records are structured: request-scoped fields bound once and merged everywhere.
• structlog vs Loguru vs standard library logging — the decision this migration assumes you have already made, with the trade-offs laid out.
• structlog JSON logging in Django — the same ProcessorFormatter bridge expressed as a Django LOGGING dictionary.
• Configuring logging with dictConfig — how to keep a declarative handler topology while swapping only the formatter.

Frequently Asked Questions