Structured JSON Logging with Request IDs in FastAPI

Key takeaways:

  • Store the correlation ID in a contextvar so it is ambient for the whole request.
  • A logging filter injects request_id into every record automatically.
  • A JSON formatter makes logs machine-parseable and filterable.
  • Add trace and span IDs so a log line links to its trace.
  • Configure the handler, filter, and formatter once at startup.

This guide builds the logging substrate that Observability and Tracing relies on, extending the request-tracing middleware.

The Problem This Solves

When an incident spans many log lines and several services, free-text logs are nearly useless — you cannot reliably isolate one request. Structured JSON logs keyed by a correlation ID turn that into a single, precise filter, and they link cleanly to traces and metrics.

Prerequisites

  • Tracing middleware that sets a correlation ID in a contextvar.
  • A log backend that ingests JSON (most do).

Step-by-Step Implementation

1. The contextvar and filter

# app/logging_setup.py
import logging
from contextvars import ContextVar

request_id_ctx: ContextVar[str] = ContextVar("request_id", default="-")


class ContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_ctx.get()   # Present on every record.
        return True

2. A JSON formatter

import json


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "level": record.levelname,
            "logger": record.name,
            "request_id": getattr(record, "request_id", "-"),
            "message": record.getMessage(),
        }
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        return json.dumps(payload)

3. Configure once at startup

def configure_logging() -> None:
    handler = logging.StreamHandler()
    handler.addFilter(ContextFilter())
    handler.setFormatter(JsonFormatter())
    root = logging.getLogger()
    root.handlers = [handler]
    root.setLevel(logging.INFO)

4. Set the ID in middleware

import uuid

from starlette.middleware.base import BaseHTTPMiddleware


class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        rid = request.headers.get("x-request-id") or str(uuid.uuid4())
        token = request_id_ctx.set(rid)
        try:
            return await call_next(request)
        finally:
            request_id_ctx.reset(token)   # Avoid leaking the value across requests.

Edge Cases and Gotchas

  • Uvicorn access logs. They use their own format; align them with your JSON formatter or log access in middleware.
  • Background tasks. A job runs outside the request context; forward the ID into the task and re-set the contextvar.
  • Sensitive data. Never log secrets or full request bodies; log identifiers and shapes.

Verification

import json


def test_logs_are_json_with_request_id(client, capsys):
    client.get("/health", headers={"x-request-id": "rid-7"})
    line = capsys.readouterr().out.strip().splitlines()[-1]
    record = json.loads(line)                 # Parses as JSON.
    assert record["request_id"] == "rid-7"