FastAPI Structured Logging with structlog and Request IDs

What structured logging is and why plain-text logs fail at scale

A log line is usually written for a human reading a terminal: logger.info(f"GET /orders/42 returned 200 in 38ms"). Structured logging treats each line as an event record with named fields. The message becomes one short key, and the data lives in its own keys: request_id, path, status_code, duration_ms.

Here is the same event both ways:

• Text: 2024-05-01 12:00:03 INFO GET /orders/42 returned 200 in 38ms
• JSON: {"event": "request_finished", "request_id": "a1b2c3", "method": "GET", "path": "/orders/42", "status_code": 200, "duration_ms": 38}

The text line breaks as soon as someone edits the wording. If "returned 200 in" becomes "-> 200 (38 ms)", every regex in your dashboards and alerts stops matching, and nothing warns you. The JSON line keeps a fixed schema. A log aggregator indexes each field directly, so a query like status_code >= 500 AND path = "/orders/*", or a p95 of duration_ms grouped by path, needs no parsing rules. A shared request_id also lets you pull every line from one request out of output from many interleaved workers.

Structured logs pay off most for teams that run several replicas or services behind a central aggregator, where nobody reads raw output line by line. A single-process CLI tool or a small internal script that someone tails locally can stay with plain text. The JSON overhead buys little there.

Copyimport logging

import structlog

# Plain text: the data is baked into the sentence, so any parser
# has to regex it back out, and rewording the message breaks that parser.
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logging.getLogger("plain").info(f"GET /orders/42 returned 200 in {38}ms")

# Structured: each call emits one event record with named fields.
structlog.configure(
    processors=[
        structlog.processors.add_log_level,                # adds "level"
        structlog.processors.TimeStamper(fmt="iso", utc=True),  # adds "timestamp"
        structlog.processors.JSONRenderer(sort_keys=True),  # one JSON object per line
    ]
)
log = structlog.get_logger()

# The message is a short, stable event name. The data lives in its own keys,
# so an aggregator can filter on status_code or compute percentiles on
# duration_ms without any parsing rules.
log.info(
    "request_finished",
    request_id="a1b2c3",
    method="GET",
    path="/orders/42",
    status_code=200,   # stays an int, so numeric queries work
    duration_ms=38,
)

# Output shape (timestamp will differ):
# {"duration_ms": 38, "event": "request_finished", "level": "info",
#  "method": "GET", "path": "/orders/42", "request_id": "a1b2c3",
#  "status_code": 200, "timestamp": "2024-05-01T12:00:03.000000Z"}

The same request event logged twice: once as an f-string line through the standard library and once as a JSON record with typed, named fields through structlog's JSONRenderer.

How the pipeline works: structlog, the stdlib bridge and contextvars

Every log call passes through two possible paths that end at one renderer. Your application code calls a structlog logger, which sends an event dict (a plain Python dictionary) through a processor chain. Each processor is a function that receives the dict and returns a modified one. Typical processors merge context, add the level and timestamp, and format exceptions. Uvicorn, SQLAlchemy, httpx and other libraries log through the standard logging module instead. They never touch structlog directly.

The bridge is structlog.stdlib.ProcessorFormatter, attached to a single handler on the root logger. It runs foreign stdlib records through a pre-chain of the same processors and hands structlog's own events to the final renderer. Both paths emit identical JSON on one stream, so the aggregator sees one schema rather than two.

Copyrequest ──► ASGI middleware ──► bind_contextvars(request_id=...)
                │                        │
                ▼                        ▼
          route handler            contextvars (per task)
           │         │                   │
   structlog log   stdlib log            │
           │         │                   │
           ▼         ▼                   │
      processor chain ◄── merge_contextvars
                │
                ▼
   ProcessorFormatter on root handler ──► JSON / console renderer ──► stdout

Per-request fields such as request_id live in contextvars. When asyncio creates a task, it copies the current context into that task. Values bound inside one request stay in that request's copy. A thread-local would fail here. A single event-loop thread interleaves many concurrent requests, so they would all share one thread-local slot. Request A's ID would then appear on request B's log lines whenever B resumed after A had written to it.

Copy  request A           request B        (one event-loop thread)
      |                   |
      v                   v
+---------------------------------------------+
| ASGI middleware                             |
| bind_contextvars(request_id=...) per task   |
+---------------------------------------------+
      |                   |
      v                   v
+---------------+   +---------------+
| ctx copy A    |   | ctx copy B    |
| request_id=A  |   | request_id=B  |
+---------------+   +---------------+
      |   route handlers and libraries log
      v                          v
+---------------------+   +------------------------+
| structlog logger    |   | stdlib logger          |
| processor chain:    |   | (uvicorn, httpx,       |
| merge_contextvars,  |   |  SQLAlchemy)           |
| level, timestamp    |   +------------------------+
+---------------------+               |
      |                               v
      |                   +------------------------+
      |                   | root handler with      |
      +------------------>| ProcessorFormatter     |
                          | (pre-chain for stdlib) |
                          +------------------------+
                                      |
                                      v
                 JSON / console renderer ---> stdout

Each request's asyncio task holds its own context copy, and both structlog and stdlib log calls reach one ProcessorFormatter on the root handler, so every line shares one schema.

Configuring structlog once for JSON in production and console output locally

That is why the configuration starts with structlog.contextvars.merge_contextvars, which reads per-task context instead of thread-local state. Put every setting in one module, such as logging_config.py, with a single configure_logging() function that you call once at startup, before the app object is created.

The shared processor chain

Build one list of processors that both structlog loggers and stdlib loggers use: merge_contextvars, add_log_level and TimeStamper(fmt="iso", utc=True). Choose the renderer from an environment variable such as LOG_FORMAT. In production, append format_exc_info and then JSONRenderer(). Locally, use ConsoleRenderer() only. It formats tracebacks itself and warns if format_exc_info runs before it.

Bridging the stdlib

End the structlog chain with ProcessorFormatter.wrap_for_formatter and attach one StreamHandler to the root logger. Give that handler a ProcessorFormatter whose foreign_pre_chain is the shared list, so records from uvicorn, SQLAlchemy or httpx get the same timestamps, levels and context. Pass log_config=None to uvicorn, or clear the handlers on uvicorn, uvicorn.error and uvicorn.access and set propagate=True on each. Otherwise uvicorn installs its own formatters.

Common misconfigurations

• Duplicate lines. These appear when uvicorn's handlers and the root handler both emit the same record.
• Escaped JSON inside event. This happens when JSONRenderer runs in the structlog chain and again in the formatter, so the second pass encodes the first pass's output as a string.
• Missing library logs. Records disappear when dictConfig keeps disable_existing_loggers=True or when the root level filters them out.

Dev and prod render through different code paths. Check output from your own code, uvicorn and at least one library in both modes.

Copy# logging_config.py: call configure_logging() once, before creating the app
import logging, os, sys
import structlog

def configure_logging() -> None:
    # Shared by structlog loggers and foreign (stdlib) records
    shared = [
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
    ]
    if os.getenv("LOG_FORMAT", "console") == "json":
        final = [structlog.processors.format_exc_info, structlog.processors.JSONRenderer()]
    else:
        final = [structlog.dev.ConsoleRenderer()]  # renders tracebacks itself

    # No renderer here: rendering happens once, in the formatter
    structlog.configure(
        processors=shared + [structlog.stdlib.ProcessorFormatter.wrap_for_formatter],
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )
    formatter = structlog.stdlib.ProcessorFormatter(
        foreign_pre_chain=shared,
        processors=[structlog.stdlib.ProcessorFormatter.remove_processors_meta, *final],
    )
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(formatter)
    root = logging.getLogger()
    root.handlers = [handler]  # replace, don't append: avoids duplicate lines
    root.setLevel(logging.INFO)

    # Route uvicorn through the root handler (or run uvicorn with log_config=None)
    for name in ("uvicorn", "uvicorn.error", "uvicorn.access"):
        lg = logging.getLogger(name)
        lg.handlers.clear()
        lg.propagate = True

A single configure_logging() that picks JSON or console rendering from LOG_FORMAT and routes structlog, uvicorn and other stdlib loggers through one ProcessorFormatter.

Implementing request ID and access-log middleware as pure ASGI

The middleware is a plain callable that takes scope, receive and send. For any scope type other than http, it passes the call through untouched. For HTTP, it first calls structlog.contextvars.clear_contextvars() so no values leak from a previous request handled on the same task. It then reads X-Request-ID and accepts it only if it is 1–64 characters drawn from [A-Za-z0-9._-]. Anything else is replaced with uuid.uuid4().hex. This blocks log injection and keeps oversized headers out of your index. Next it binds request_id, method, path and the client address. It wraps send so that on http.response.start it records the status code and appends the request ID to the response headers. Finally, in a finally block, it emits one access event with status_code and duration_ms computed from time.perf_counter(), which is monotonic and immune to wall-clock jumps.

That event replaces uvicorn's access log. Start uvicorn with --no-access-log so each request produces one line, not two with different shapes.

Pure ASGI is preferable to BaseHTTPMiddleware. That class has historically run the endpoint in a separate task, so contextvars set downstream did not propagate back. It also had known problems with streaming responses and background tasks. A raw ASGI callable has none of these indirections.

Copyimport re, time, uuid
import structlog

logger = structlog.get_logger("access")
# Accept only short IDs from a safe alphabet, which blocks log injection.
_VALID_ID = re.compile(r"[A-Za-z0-9._-]{1,64}")

class RequestIDMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":  # lifespan and websocket pass through untouched
            return await self.app(scope, receive, send)
        structlog.contextvars.clear_contextvars()  # drop leftovers from a prior request
        raw = dict(scope["headers"]).get(b"x-request-id", b"").decode("latin-1")
        request_id = raw if _VALID_ID.fullmatch(raw) else uuid.uuid4().hex
        client = scope.get("client")  # (host, port) tuple, or None
        structlog.contextvars.bind_contextvars(
            request_id=request_id, method=scope["method"], path=scope["path"],
            client=f"{client[0]}:{client[1]}" if client else None)
        status_code = 500  # stays 500 if the app raises before it starts a response
        start = time.perf_counter()  # monotonic, so wall-clock jumps cannot skew it

        async def send_wrapper(message):
            nonlocal status_code
            if message["type"] == "http.response.start":
                status_code = message["status"]
                headers = list(message.get("headers", []))
                headers.append((b"x-request-id", request_id.encode("ascii")))
                message = {**message, "headers": headers}  # echo the ID to the caller
            await send(message)

        try:
            await self.app(scope, receive, send_wrapper)
        finally:  # exactly one access event per request, even when the app raises
            logger.info("access", status_code=status_code,
                        duration_ms=round((time.perf_counter() - start) * 1000, 2))

# Usage: app.add_middleware(RequestIDMiddleware), then run: uvicorn main:app --no-access-log

A pure ASGI middleware validates or generates X-Request-ID, binds request context with structlog, echoes the ID in the response, and emits a single access event that replaces uvicorn's access log.

Logging exceptions and propagating context beyond the request coroutine

Inside a handler, call logger.exception("payment_failed") from an except block, or pass exc_info=True to any level. Either way the exception travels as a value on the event rather than as text glued to the message. With format_exc_info or dict_tracebacks in the processor chain, the JSON renderer emits the traceback as an exception field, so a single log record holds the error, its stack and the bound request_id. Your log backend can then group and filter errors by those fields.

Register a catch-all handler for Exception that does two things: log with exc_info, and return a 500 body containing request_id. A user who reports that ID gives you a direct lookup key. Starlette runs the catch-all handler in its outermost error middleware, which can execute after your request-ID middleware has cleared its contextvars. For that reason, store the ID on scope["state"] as well and read it from request.state in the handler.

Context propagates in two cases:

• asyncio tasks copy the current context when you create them.
• Sync endpoints and dependencies run in Starlette's threadpool with a copied context, so their logs carry request_id.

Context does not propagate in these cases:

• Thread pools you manage yourself. loop.run_in_executor does not copy context. Wrap the call in contextvars.copy_context().run.
• Separate processes and message queues. Put request_id in the message payload and bind it again in the consumer.

BackgroundTasks run after the response is sent, and how they interact with your middleware's cleanup depends on how that middleware is built. Write a test that asserts request_id is present on a background task's log line rather than assuming it is.

Testing the setup with pytest and keeping sensitive data out of logs

Testing logging means asserting on structured fields, not matching rendered strings. structlog.testing.capture_logs() is the quickest helper, but it temporarily replaces your processor chain, so merge_contextvars never runs and request_id will not appear. To exercise the real pipeline, append a structlog.testing.LogCapture instance to your configured processors in a fixture. If you route through the stdlib bridge, you can also read caplog.records and parse each message.

Asserting request isolation

Send two requests at once with asyncio.gather over an httpx.AsyncClient bound to the app through ASGITransport. Then assert three things:

• Each response's X-Request-ID header matches the request_id on its own access-log entry.
• The two IDs differ.
• No log entry carries the other request's ID.

This test fails if middleware binds context to a shared object instead of a contextvar, which is a common regression.

Redaction as a backstop

Add a processor before the renderer that walks the event dict and replaces values under keys such as authorization, password, token, cookie and api_key with a fixed placeholder. Match keys case-insensitively and recurse into nested dicts. Treat this as a safety net, not the policy. The policy is that middleware never binds full request bodies or header maps, because any key the deny-list does not anticipate leaks.

A fixed field schema

Define an allow-list of access-log fields, for example method, route template, status, duration and request ID, and assert in a test that entries contain nothing else. Log the route template (/users/{id}) rather than the raw path. Leave out query strings, user agents and payloads: they carry PII and unbounded cardinality, which hurts indexing cost and query performance in most log backends.

When to avoid it, alternatives, and impact on cost and engineering time

All of this has a price. Bridging stdlib loggers into structlog means maintaining two configurations that must agree on processors and output format. Every log call builds a dict, runs it through processors and serializes it to JSON. Profile that cost on your own hot paths instead of assuming it is negligible. Stale context is the subtler failure: if a contextvar is not cleared, or a worker thread inherits a stale copy, one request's request_id appears on another request's lines.

JSON also makes logs larger, and aggregators bill by volume:

Illustrative estimate (assumed numbers, not a benchmark): daily GB = requests/day × lines/request × bytes/line ÷ 10⁹. With 5,000,000 requests, 4 lines and 600 bytes, that is 12 GB/day. At an assumed $0.50/GB, it costs $6/day, or about $180 a month.

A request ID does not replace tracing. It identifies one hop, while a trace links spans across services. When tracing is present, log trace_id and span_id next to request_id so you can move from a log line to its trace.

The return on this work comes during incidents. With a consistent schema, an on-call engineer filters on one request_id and gets that request's full history. Without one, they grep several services and line up events by timestamp, which breaks down under concurrency and clock skew.

How OneEnv can help

Disclosure: OneEnv is the product behind this blog.

OneEnv (oneenv.io) is a workspace-based governance platform for the things services share — environment variables, database structure, API routes and validation rules. A change to a shared value is staged on a branch, reviewed once, and then every affected service explicitly approves or rejects it before it goes live, so you can see who a change touches before it ships instead of finding out in production.

Applied to fastapi structured logging, this is where it helps:

Saves time

• Every service that inherits a shared value is listed in one place, so nobody searches repos and chats to find out who is affected.
• One review per branch replaces coordinating each team separately; each service still keeps its own approve/reject.
• Changing a shared variable is one edit and one review instead of editing every service's .env by hand.

Saves cost

• Fewer production incidents and rollbacks caused by a shared value, route or column changing under a consumer unannounced.
• Less duplicated configuration to maintain: one shared definition inherited by many services.
• A wrong or stale value is caught at review time by the services it affects, not by a failed deploy.
• OneEnv has a free plan (1 workspace, unlimited services and members, full governance), so trying it costs nothing.

These are qualitative benefits, not measured results — how much you save depends on how many services share the value.

Want to try it? Start at oneenv.io.