FastAPI Backend Development Bootcamp · Lektion

Strukturiertes JSON-Logging und Korrelations-IDs

Erzeugen Sie strukturierte Logs mit requestbezogenen Korrelations-IDs, die asynchrone Grenzen und Services hinweg erhalten bleiben.

Strukturiertes JSON-Logging und Korrelations-IDs ist eine kostenlose FastAPI Backend Development Bootcamp-Lektion auf CoddyKit. Dies ist Lektion 1 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des FastAPI Backend Development Bootcamp-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der FastAPI Backend Development Bootcamp-Kurs umfasst insgesamt 4 Lektionen.

Teile dieser Lektion wurden noch nicht übersetzt und werden auf Englisch angezeigt.

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.

Häufig gestellte Fragen

Ist die Lektion „Strukturiertes JSON-Logging und Korrelations-IDs“ kostenlos?

Ja — der vollständige Text von „Strukturiertes JSON-Logging und Korrelations-IDs“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des FastAPI Backend Development Bootcamp-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der FastAPI Backend Development Bootcamp-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Strukturiertes JSON-Logging und Korrelations-IDs“?

Erzeugen Sie strukturierte Logs mit requestbezogenen Korrelations-IDs, die asynchrone Grenzen und Services hinweg erhalten bleiben. Du übst FastAPI Backend Development Bootcamp mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um FastAPI Backend Development Bootcamp zu starten?

Keine Vorkenntnisse erforderlich. FastAPI Backend Development Bootcamp auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 1 von 4.

Wie lange dauert die Lektion „Strukturiertes JSON-Logging und Korrelations-IDs“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser FastAPI Backend Development Bootcamp-Lektion Code schreiben und ausführen?

Ja. Jede FastAPI Backend Development Bootcamp-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

Strukturiertes JSON-Logging und Korrelations-IDs
Verteiltes Tracing mit OpenTelemetry
Prometheus-Metriken und RED-/USE-Dashboards
Alarmierung zu SLOs und Error Budgets

← Zurück zu FastAPI Backend Development Bootcamp