How to Add Structured Logging to a FastAPI App: JSON Logs, Request IDs, and Correlation

print() is a fine logger right up until the moment you have two of anything — two workers, two requests in flight, two servers behind a load balancer. Then the console becomes a braid of interleaved lines from unrelated requests, with no way to tell which line belongs to which, and the first real production incident turns into archaeology. Structured logging is the fix, and it's less work than the mess it replaces: emit each log as a JSON object instead of a sentence, stamp every line with a request ID, and make that ID propagate automatically so a single query pulls back the entire story of one request. This is a practical, vendor-neutral guide to doing exactly that in FastAPI — standard-library Python, one small middleware, and one idea (contextvars) that ties it all together.

Why JSON Logs, Not Sentences

A human-readable log line — User 4821 updated order 9910 in 240ms — reads fine to you and is nearly useless to a machine. The moment you want to answer “show me every request slower than 500ms for user 4821 yesterday,” you're writing regexes against prose. A structured log emits the same event as data:

{"timestamp": "2026-07-06T14:03:11Z", "level": "INFO", "logger": "app.orders",
 "message": "order updated", "user_id": 4821, "order_id": 9910, "duration_ms": 240,
 "request_id": "6f1c…"}

Now every field is queryable. Any log aggregator — or a local jq pipe — can filter on user_id, sort by duration_ms, or group by level without parsing English. The shift is small and total: you stop formatting facts into a string and start attaching them as fields. Everything below is in service of that one move.

Step 1: Emit JSON From the Standard Library

You don't need a framework for this. Python's built-in logging already separates what you log from how it's rendered — the rendering lives in a Formatter, so a JSON formatter is the entire lift. Configure it once at startup:

import json, logging, sys

class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%SZ"),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "request_id": getattr(record, "request_id", "-"),
        }
        # anything passed via logger.info("...", extra={...}) rides along
        for key, value in getattr(record, "context", {}).items():
            payload[key] = value
        if record.exc_info:
            payload["exception"] = self.formatException(record.exc_info)
        return json.dumps(payload, default=str)

def configure_logging(level: str = "INFO") -> None:
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JsonFormatter())
    root = logging.getLogger()
    root.handlers.clear()          # drop Uvicorn's default text handler
    root.addHandler(handler)
    root.setLevel(level)

Call configure_logging() before you create the app, log to stdout, and stop there. In a container, stdout is the log — the orchestrator collects it, and rotation, shipping, and retention are the platform's job, not your app's. Writing to files from inside the process is a habit worth dropping the day you containerize. Reach for a library like structlog if you want its ergonomics, but understand you don't need it: the standard library does structured output in the twenty lines above.

Step 2: Give Every Request an ID

The single most useful field in the whole scheme is the one that ties a request's log lines together. Generate a unique ID at the edge of every request, and — critically — honor one that's already there: if a proxy, gateway, or calling service sent an X-Request-ID, reuse it so the trail stitches across service boundaries instead of restarting at yours.

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 uuid.uuid4().hex
        token = request_id_var.set(rid)          # see Step 3
        try:
            response = await call_next(request)
        finally:
            request_id_var.reset(token)          # don't leak the ID to the next request
        response.headers["X-Request-ID"] = rid   # echo it so the caller can correlate too
        return response

app.add_middleware(RequestIdMiddleware)

Echoing the ID back in the response header is a small courtesy that pays off constantly: when a user reports “it failed at 2:04,” the ID is right there in their browser's network tab, and you paste it into one query instead of guessing at timestamps.

Step 3: Make the ID Propagate by Itself

Here's the part that separates logging that helps from logging you have to babysit. You do not want to thread a request_id argument through every function call down to the query layer. Instead, store it in a contextvars.ContextVar — a variable that's naturally scoped to the current async task — and let a logging Filter stamp it onto every record automatically:

import contextvars, logging

request_id_var: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")

class RequestIdFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_var.get()   # injected into JsonFormatter above
        return True

# attach the filter once, wherever you configured the handler
handler.addFilter(RequestIdFilter())

That's the whole trick. The middleware sets the contextvar at the start of the request; the filter reads it for every log line emitted anywhere in that request's call stack; the formatter writes it out. A logger.info("order updated") buried three layers deep in a service function now carries the right request_id without knowing the ID exists. Because a ContextVar is isolated per async task, two requests handled concurrently never bleed IDs into each other.

Step 4: Log the Request Itself — Once

With context wired up, add one deliberate log line per request: a single completion event that captures method, path, status, and duration. It's the backbone of your observability — the record you group, count, and alert on — and it belongs in the same middleware:

import time, logging

log = logging.getLogger("app.access")

class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        rid = request.headers.get("X-Request-ID") or uuid.uuid4().hex
        token = request_id_var.set(rid)
        start = time.perf_counter()
        try:
            response = await call_next(request)
        finally:
            elapsed_ms = round((time.perf_counter() - start) * 1000, 1)
            log.info("request completed", extra={"context": {
                "method": request.method,
                "path": request.url.path,
                "status": locals().get("response").status_code if "response" in locals() else 500,
                "duration_ms": elapsed_ms,
            }})
            request_id_var.reset(token)
        response.headers["X-Request-ID"] = rid
        return response

One structured line per request, and your dashboards write themselves: request rate is a count, latency is a percentile over duration_ms, error rate is a filter on status >= 500. Resist the urge to also log the start of every request — it doubles your volume to tell you something the completion line already implies. Log the outcome, richly, once.

Step 5: Use Levels Like They Mean Something

Log levels are a routing decision, not decoration — they're how you and your alerting tell “someone typed a bad password” apart from “the database is down.” A working convention:

• DEBUG — developer detail (the SQL it built, the branch it took). Off in production; invaluable when you flip it on.
• INFO — things happened as designed: a request completed, an order shipped, a job ran. Your default, and the completion line from Step 4.
• WARNING — degraded but handled: a retry succeeded, a cache missed, a deprecated path got hit. Someone should look eventually, no one gets paged.
• ERROR — this request failed and a human needs to know: an unhandled exception, a downstream call that exhausted its retries.
• CRITICAL — the process itself is in trouble. Rare by design; when it fires, it means it.

The most common mistake is logging expected conditions at ERROR. A user submitting an invalid form is a 400 and an INFO at most — it is not an error in your system, and if it pages you at 3 a.m., you'll soon be ignoring the channel that also carries the real fire. Reserve ERROR for things that are genuinely your bug or a genuine outage, and your alerts stay trustworthy.

The One Gotcha: Background Work

Context propagation has a sharp edge worth knowing before it bites you. A ContextVar follows the current async task, so it flows naturally into anything you await and into FastAPI's BackgroundTasks. But the moment you hand work to a separate executor — a thread pool, a process, a Celery or RQ worker on another machine — that new context starts empty, and your logs there show request_id: "-". The fix is to pass the ID across the boundary explicitly and re-seat it on the other side:

rid = request_id_var.get()
enqueue_job(process_export, export_id=42, request_id=rid)   # carry it across the wire

def process_export(export_id: int, request_id: str) -> None:
    request_id_var.set(request_id)     # re-seat context in the worker
    log.info("export started", extra={"context": {"export_id": export_id}})

Do that, and a request that kicks off async work stays traceable end to end — the web log and the worker log share one ID, and “where did this export go?” is one query across both.

The Bottom Line

Structured logging isn't a library you install or a vendor you sign up for — it's a handful of decisions you make once and then benefit from on every incident afterward. Emit JSON so your logs are data, not prose. Mint a request ID at the edge and honor one that arrives. Park it in a ContextVar and let a filter stamp it on every line, so correlation costs you nothing at the call site. Log each request's outcome once, richly, and use levels that mean what they say. The payoff is the difference between grepping a wall of interleaved text and running one query that returns the complete, ordered story of a single request — which is the only thing you ever actually wanted from a log.