Correlation middleware

When a user reports an error, you want to find every log line that request produced. A single request id in every line makes that a one-command search. The trick is getting the id to every log call without threading it through every function. ContextVar is the clean answer.

Request id propagation

Middleware sets a ContextVar. Log filter reads it on every record.

middleware.py
python
import logging
import uuid
from contextvars import ContextVar
from typing import Optional

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

request_id_ctx: ContextVar[Optional[str]] = ContextVar("request_id", default=None)

class RequestIDFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_ctx.get()
        return True

class RequestIDMiddleware(BaseHTTPMiddleware):
    HEADER = "X-Request-ID"

    async def dispatch(self, request: Request, call_next) -> Response:
        incoming = request.headers.get(self.HEADER)
        rid = incoming or str(uuid.uuid4())
        token = request_id_ctx.set(rid)
        try:
            response = await call_next(request)
        finally:
            request_id_ctx.reset(token)
        response.headers[self.HEADER] = rid
        return response

ContextVar is scoped to the current asyncio task, so concurrent requests never share ids. The middleware accepts an incoming X-Request-ID (preserving ids from upstream systems) or generates a new one, and echoes it back on the response so clients can include it in bug reports.

middleware.py
python
def install_request_id_log_filter() -> None:
    """Attach the RequestIDFilter to the root logger so every record carries it."""
    root = logging.getLogger()
    if not any(isinstance(f, RequestIDFilter) for f in root.filters):
        root.addFilter(RequestIDFilter())

One attach at startup and every logger in the app picks up the request id on every record, with no code changes at call sites. The idempotency guard means repeated imports in tests do not stack duplicate filters.

Quiz: Quiz

Loading practice…