FastAPI Backend Development Bootcamp · Lección

Registro JSON estructurado e identificadores de correlación

Emita logs estructurados con identificadores de correlación asociados a la solicitud que se conserven entre límites asíncronos y servicios.

Registro JSON estructurado e identificadores de correlación es una lección gratuita de FastAPI Backend Development Bootcamp en CoddyKit. Esta es la lección 1 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de FastAPI Backend Development Bootcamp, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de FastAPI Backend Development Bootcamp incluye 4 lecciones en total.

Partes de esta lección aún no han sido traducidas y se muestran en inglés.

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.

Preguntas frecuentes

¿La lección «Registro JSON estructurado e identificadores de correlación» es gratis?

Sí — el texto completo de «Registro JSON estructurado e identificadores de correlación» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de FastAPI Backend Development Bootcamp, actualiza a CoddyKit PRO. El curso de FastAPI Backend Development Bootcamp incluye 4 lecciones en total.

¿Qué aprenderé en «Registro JSON estructurado e identificadores de correlación»?

Emita logs estructurados con identificadores de correlación asociados a la solicitud que se conserven entre límites asíncronos y servicios. Practicas FastAPI Backend Development Bootcamp con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.

¿Necesito experiencia previa para empezar FastAPI Backend Development Bootcamp?

No se requiere experiencia previa. FastAPI Backend Development Bootcamp en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 1 de 4.

¿Cuánto tiempo toma la lección «Registro JSON estructurado e identificadores de correlación»?

La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.

¿Puedo escribir y ejecutar código en esta lección de FastAPI Backend Development Bootcamp?

Sí. Cada lección de FastAPI Backend Development Bootcamp incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.

Todas las lecciones de este curso

Registro JSON estructurado e identificadores de correlación
Trazabilidad distribuida con OpenTelemetry
Métricas de Prometheus y dashboards RED/USE
Alertas sobre SLO y presupuestos de error

← Volver a FastAPI Backend Development Bootcamp