0Pricing
FastAPI Backend Development Bootcamp · Lesson

Structured JSON Logging and Correlation IDs

Emit structured logs with request-scoped correlation IDs that survive across async boundaries and services.

Structured JSON Logging and Correlation IDs is a free FastAPI Backend Development Bootcamp lesson on CoddyKit — lesson 1 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the FastAPI Backend Development Bootcamp learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

Why Structured Logs

In production, logs are data, not prose. A line like User 42 failed login from 10.0.0.3 reads fine to a human but is painful for machines: you cannot reliably filter, aggregate, or alert on it.

Structured logging emits each event as a JSON object with stable, queryable fields:

  • timestamp, level, message
  • request_id / correlation_id
  • context such as user_id, path, status_code, duration_ms

Log aggregators (Loki, Elasticsearch, Datadog) then index those fields so you can run queries like level=ERROR AND path=/checkout.

A JSON Log in One Line

The simplest structured log is just a dictionary serialized to JSON on one line. One JSON object per line is the JSON Lines (NDJSON) format that virtually every log shipper understands.

This standalone example shows the shape we are aiming for. Notice the fields are flat and named consistently.

import json
import time

def log(level, message, **fields):
    record = {
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "level": level,
        "message": message,
        **fields,
    }
    print(json.dumps(record))

log("INFO", "request completed", path="/checkout", status_code=200, duration_ms=42)
log("ERROR", "db timeout", path="/orders", correlation_id="abc-123")

A Custom JSON Formatter

Rolling your own print(json.dumps(...)) bypasses Python's logging module, losing levels, handlers, and library logs. Instead, plug a JSON formatter into the standard logging stack.

A formatter's job is to turn a LogRecord into a string. Here we return JSON. record.__dict__ carries any extra={...} fields you pass at the call site.

import json
import logging

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

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])

logging.getLogger("app").info("service started", extra={"port": 8000})

The Correlation ID Problem

A single user request often fans out: API handler -> service layer -> database call -> outbound HTTP call to another service. If each log line is anonymous, you cannot stitch them back into one story.

A correlation ID (a.k.a. request ID or trace ID) is a unique value generated once per inbound request and attached to every log line produced while handling it. Then correlation_id=abc-123 retrieves the full timeline across functions and even across services.

The challenge: how do you make that ID available deep in the call stack without threading it through every function argument?

ContextVar: Request-Scoped State

The clean answer is contextvars.ContextVar. Unlike a global variable, a ContextVar holds a value that is isolated per logical execution context and, crucially, propagates correctly across async awaits.

Each concurrent request runs in its own context, so setting the correlation ID in one request never leaks into another, even when many run interleaved on the same event loop.

import asyncio
from contextvars import ContextVar

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

async def handle(name, cid):
    correlation_id.set(cid)
    await asyncio.sleep(0.01)
    # value survives the await and stays isolated per task
    print(name, "->", correlation_id.get())

async def main():
    await asyncio.gather(
        handle("req-A", "aaa"),
        handle("req-B", "bbb"),
    )

asyncio.run(main())

Injecting the ID via a Log Filter

To get the correlation ID onto every log line automatically, attach a logging.Filter that reads the ContextVar and copies it onto the record. A filter runs for every record passing through the handler, so no call site has to remember to pass the ID.

The formatter then reads record.correlation_id like any other field.

import json
import logging
from contextvars import ContextVar

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

class CorrelationFilter(logging.Filter):
    def filter(self, record):
        record.correlation_id = correlation_id.get()
        return True

class JsonFormatter(logging.Formatter):
    def format(self, record):
        return json.dumps({
            "level": record.levelname,
            "message": record.getMessage(),
            "correlation_id": getattr(record, "correlation_id", "-"),
        })

h = logging.StreamHandler()
h.addFilter(CorrelationFilter())
h.setFormatter(JsonFormatter())
logging.basicConfig(level=logging.INFO, handlers=[h])

correlation_id.set("abc-123")
logging.getLogger("app").info("order placed")

FastAPI Middleware to Set the ID

In FastAPI, the right place to establish the correlation ID is an HTTP middleware, which wraps every request. The pattern:

  • Read an incoming X-Request-ID / X-Correlation-ID header if a caller (gateway, upstream service) already set one.
  • Otherwise generate a fresh UUID.
  • Store it in the ContextVar so all downstream logs pick it up.
  • Echo it back in the response header so clients can report it in bug reports.

This is framework code that needs a running server, so it is illustrative rather than runnable.

import uuid
from fastapi import FastAPI, Request
from contextvars import ContextVar

correlation_id: ContextVar[str] = ContextVar("correlation_id", default="-")
app = FastAPI()

@app.middleware("http")
async def correlation_middleware(request: Request, call_next):
    cid = request.headers.get("X-Request-ID") or str(uuid.uuid4())
    token = correlation_id.set(cid)
    try:
        response = await call_next(request)
    finally:
        correlation_id.reset(token)
    response.headers["X-Request-ID"] = cid
    return response

Why reset() with a Token Matters

Notice token = correlation_id.set(cid) followed by correlation_id.reset(token) in a finally block. The token restores the previous value when the request ends.

Under an ASGI server, worker tasks and contexts can be reused. Resetting prevents a stale ID from a finished request from bleeding into a later one that forgot to set its own. Always pair set() with reset() in middleware, and do it in finally so it runs even when the handler raises.

from contextvars import ContextVar

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

print(cv.get())          # -
token = cv.set("req-1")
print(cv.get())          # req-1
cv.reset(token)
print(cv.get())          # back to -

Surviving Background Tasks and Threads

ContextVar propagates automatically across await within the same task, but a value does not automatically follow work you push to another thread (for example run_in_executor or blocking DB drivers).

To carry the context across a thread boundary, capture it with contextvars.copy_context() and run the callable inside that copy. asyncio already does this for create_task; you must do it manually for raw executors.

import contextvars
from concurrent.futures import ThreadPoolExecutor

cid = contextvars.ContextVar("cid", default="-")

def work():
    return cid.get()

cid.set("trace-9")
ctx = contextvars.copy_context()
with ThreadPoolExecutor() as pool:
    # ctx.run carries the ContextVar value into the worker thread
    result = pool.submit(ctx.run, work).result()

print("in thread:", result)  # trace-9

Propagating Across Services

A correlation ID is only useful end-to-end if it crosses service boundaries. When your FastAPI service calls another service, forward the ID as an HTTP header so the downstream logs share the same value.

Read it from the ContextVar and inject it into every outbound client call. The receiving service's middleware reads that header instead of generating a new ID, so one ID spans the whole call chain.

import httpx
from contextvars import ContextVar

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

async def call_downstream(url: str):
    headers = {"X-Request-ID": correlation_id.get()}
    async with httpx.AsyncClient() as client:
        resp = await client.get(url, headers=headers)
        return resp.json()

Putting It Together with structlog

Rather than hand-build formatters, many teams use structlog, which composes a pipeline of processors and renders JSON at the end. A processor can pull the correlation ID from the ContextVar and merge it into every event automatically.

The benefits compound: consistent JSON output, easy per-event context binding via logger.bind(...), and clean integration with the stdlib logging module so library logs are captured too.

import structlog
from contextvars import ContextVar

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

def add_correlation_id(logger, method_name, event_dict):
    event_dict["correlation_id"] = correlation_id.get()
    return event_dict

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

correlation_id.set("abc-123")
log = structlog.get_logger()
log.info("checkout_completed", amount=49.9, currency="EUR")

Quick Check

Test your understanding of correlation ID propagation in async FastAPI services.

Recap

You built request-scoped, structured logging for FastAPI:

  • Structured JSON logs via a custom logging.Formatter (or structlog) make logs queryable.
  • Correlation IDs stitch every log line of one request together across functions and services.
  • contextvars.ContextVar holds the ID with per-request isolation and survives await boundaries.
  • A logging filter injects the ID onto every record so no call site must remember it.
  • FastAPI middleware reads X-Request-ID or generates a UUID, then pairs set() with reset(token) in finally.
  • Carry context into threads with copy_context() and across services by forwarding the ID header.

The result: one ID, queried in your log aggregator, reveals the complete journey of any request.

Frequently asked questions

Is the “Structured JSON Logging and Correlation IDs” lesson free?

Yes — the full text of “Structured JSON Logging and Correlation IDs” is free to read here on the web, and the FastAPI Backend Development Bootcamp course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the FastAPI Backend Development Bootcamp course, upgrade to CoddyKit PRO.

What will I learn in “Structured JSON Logging and Correlation IDs”?

Emit structured logs with request-scoped correlation IDs that survive across async boundaries and services. You practise FastAPI Backend Development Bootcamp with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start FastAPI Backend Development Bootcamp?

No prior experience is required. FastAPI Backend Development Bootcamp on CoddyKit is structured for beginners through advanced learners; this is — lesson 1 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Structured JSON Logging and Correlation IDs” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this FastAPI Backend Development Bootcamp lesson?

Yes. Every FastAPI Backend Development Bootcamp lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

  1. Structured JSON Logging and Correlation IDs
  2. Distributed Tracing with OpenTelemetry
  3. Prometheus Metrics and RED/USE Dashboards
  4. Alerting on SLOs and Error Budgets
← Back to FastAPI Backend Development Bootcamp