0Pricing
FastAPI Backend Development Bootcamp · Aula

Logging JSON estruturado e IDs de correlação

Emita logs estruturados com IDs de correlação vinculados ao escopo da requisição que persistam entre limites assíncronos e serviços.

Logging JSON estruturado e IDs de correlação é uma aula grátis de FastAPI Backend Development Bootcamp no CoddyKit. Esta é a aula 1 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de FastAPI Backend Development Bootcamp, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de FastAPI Backend Development Bootcamp inclui 4 aulas no total.

Partes desta aula ainda não foram traduzidas e aparecem em 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.

Perguntas Frequentes

A aula “Logging JSON estruturado e IDs de correlação” é grátis?

Sim — o texto completo de “Logging JSON estruturado e IDs de correlação” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de FastAPI Backend Development Bootcamp, atualize para CoddyKit PRO. O curso de FastAPI Backend Development Bootcamp inclui 4 aulas no total.

O que vou aprender em “Logging JSON estruturado e IDs de correlação”?

Emita logs estruturados com IDs de correlação vinculados ao escopo da requisição que persistam entre limites assíncronos e serviços. Você pratica FastAPI Backend Development Bootcamp com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.

Preciso ter experiência prévia para começar FastAPI Backend Development Bootcamp?

Nenhuma experiência prévia é necessária. FastAPI Backend Development Bootcamp no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 1 de 4.

Quanto tempo leva a aula “Logging JSON estruturado e IDs de correlação”?

A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.

Posso escrever e executar código nesta aula de FastAPI Backend Development Bootcamp?

Sim. Cada aula de FastAPI Backend Development Bootcamp inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.

Todas as aulas deste curso

  1. Logging JSON estruturado e IDs de correlação
  2. Rastreamento distribuído com OpenTelemetry
  3. Métricas do Prometheus e painéis RED/USE
  4. Alertas sobre SLOs e orçamentos de erros
← Voltar para FastAPI Backend Development Bootcamp