import contextvars
import uuid
_correlation_id: contextvars.ContextVar[str] = contextvars.ContextVar(
"correlation_id", default="-"
)
def set_correlation_id(value):
return _correlation_id.set(value)
def reset_correlation_id(token):
_correlation_id.reset(token)
def get_correlation_id():
return _correlation_id.get()
def new_correlation_id():
return uuid.uuid4().hex
from starlette.middleware.base import BaseHTTPMiddleware
from correlation import (
set_correlation_id,
reset_correlation_id,
get_correlation_id,
new_correlation_id,
)
HEADER_NAME = "X-Request-ID"
class CorrelationIdMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
incoming = request.headers.get(HEADER_NAME)
cid = incoming or new_correlation_id()
token = set_correlation_id(cid)
try:
response = await call_next(request)
finally:
reset_correlation_id(token)
response.headers[HEADER_NAME] = get_correlation_id() if incoming else cid
return response
import logging
from correlation import get_correlation_id
class CorrelationIdFilter(logging.Filter):
def filter(self, record):
record.correlation_id = get_correlation_id()
return True
def configure_logging(level=logging.INFO):
handler = logging.StreamHandler()
handler.addFilter(CorrelationIdFilter())
handler.setFormatter(
logging.Formatter(
"%(asctime)s %(levelname)s [cid=%(correlation_id)s] "
"%(name)s: %(message)s"
)
)
root = logging.getLogger()
root.handlers = [handler]
root.setLevel(level)
import logging
from fastapi import FastAPI
from middleware import CorrelationIdMiddleware
from logging_config import configure_logging
logger = logging.getLogger("orders")
app = FastAPI()
app.add_middleware(CorrelationIdMiddleware)
@app.on_event("startup")
async def _startup():
configure_logging()
async def load_order(order_id):
# nested call logs the same cid without being passed anything
logger.info("fetching order %s from repository", order_id)
return {"id": order_id, "status": "paid"}
@app.get("/orders/{order_id}")
async def get_order(order_id: str):
logger.info("handling order request")
order = await load_order(order_id)
logger.info("order resolved with status %s", order["status"])
return order
This snippet shows how to trace a single request across every layer of an async FastAPI service by threading a correlation ID through a contextvars.ContextVar instead of passing it down as a function argument. The core problem is that in an async application, many coroutines run interleaved on the same event loop, so a plain module-level global would be shared and clobbered between concurrent requests. contextvars solves this by giving each logical execution context — including tasks spawned per request — its own isolated copy of the variable, which makes it the correct primitive for request-scoped state under asyncio.
In correlation.py, the ContextVar named _correlation_id holds the current request's ID with a safe default. set_correlation_id returns a Token so the value can be restored later, and get_correlation_id is the read side used everywhere else. The lightweight helpers keep the rest of the codebase free of any knowledge of how the ID is stored, so the storage mechanism could change without touching call sites.
middleware.py is where the ID enters the system. CorrelationIdMiddleware reads an incoming X-Request-ID header for propagation across services, falling back to a fresh uuid4 when a client did not supply one. It sets the value with set_correlation_id, and crucially resets it in a finally block using the returned token so the context does not leak into whatever runs next on that worker. The same ID is echoed back on the response header so callers and load balancers can correlate too.
logging_config.py wires the ID into every log line. The CorrelationIdFilter pulls the current value via get_correlation_id and attaches it as a record attribute, letting the Formatter include %(correlation_id)s without any per-call plumbing. Because the filter reads the ContextVar at emit time, deeply nested service and repository code logs the right ID automatically.
app.py ties it together: the middleware is registered, logging is configured at startup, and the handler and a nested service call both log without ever mentioning the correlation ID. The trade-off is that contextvars state is implicit, so care must be taken with manually created tasks, which do copy the context at creation time, and with thread pools, which do not — a common pitfall when offloading blocking work.
Related snips
package api
import (
"net/http"
"runtime/debug"
)Expose build metadata for debugging deploys
use tracing::{info, instrument};
#[instrument]
fn process_request(user_id: u64) {
info!(user_id, "Processing request");
// Work happens heretracing for structured logging and distributed tracing
# Grafana provisioning: datasources
# /etc/grafana/provisioning/datasources/prometheus.yml
apiVersion: 1
datasources:
- name: Prometheus
type: prometheusGrafana dashboards as code with JSON provisioning
import React from "react";
type FallbackProps = {
error: Error;
reset: () => void;
};React Error Boundary + error reporting hook
import type { IncomingMessage, ServerResponse } from "http";
const MIN_BYTES = 1024;
const INCOMPRESSIBLE = /^(image|video|audio)\/|application\/(zip|gzip|x-brotli|pdf|octet-stream)/i;Response compression (only when it helps)
import pino, { Logger } from 'pino';
import { AsyncLocalStorage } from 'node:async_hooks';
export interface Store {
requestId: string;
logger: Logger;Request ID + structured logging (Express + pino)
Share this code
Here's the card — post it anywhere.
