FastAPI request ids in structlog contextvars, tested with capture_logs: working versions (tested 2026-10-15)

fastapi==0.143.1
structlog==26.1.0
httpx==0.28.1
pytest==9.1.1

import re
import uuid

import structlog
from fastapi import FastAPI
from structlog.contextvars import bind_contextvars, clear_contextvars, merge_contextvars

structlog.configure(processors=[merge_contextvars, structlog.processors.JSONRenderer()])
log = structlog.get_logger()

SAFE_ID = re.compile(r"[A-Za-z0-9-]{1,64}")


class RequestIdMiddleware:
    """Pure ASGI middleware: bindings happen in the same task as the endpoint,
    and the header is added on http.response.start."""

    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            return await self.app(scope, receive, send)

        # Never trust the raw header: reject anything that could break logs or headers.
        incoming = dict(scope["headers"]).get(b"x-request-id", b"").decode("latin-1")
        rid = incoming if SAFE_ID.fullmatch(incoming) else uuid.uuid4().hex

        async def send_with_header(message):
            if message["type"] == "http.response.start":
                message.setdefault("headers", []).append((b"x-request-id", rid.encode()))
            await send(message)

        bind_contextvars(request_id=rid)
        try:
            await self.app(scope, receive, send_with_header)
        finally:
            # Without this, ids leak into whatever runs next on the same context.
            clear_contextvars()


app = FastAPI()
app.add_middleware(RequestIdMiddleware)


@app.get("/hello")
async def hello():
    log.info("hello_called")
    return {"ok": True}


@app.get("/boom")
async def boom():
    log.info("about_to_fail")
    raise RuntimeError("boom")

import re

from fastapi.testclient import TestClient
from structlog.contextvars import merge_contextvars
from structlog.testing import capture_logs

from app import app

# raise_server_exceptions=False so unhandled errors come back as 500 responses.
client = TestClient(app, raise_server_exceptions=False)


def test_echoes_incoming_request_id():
    r = client.get("/hello", headers={"X-Request-ID": "abc-123"})
    assert r.status_code == 200
    assert r.headers["x-request-id"] == "abc-123"


def test_generates_id_when_missing():
    r = client.get("/hello")
    assert re.fullmatch(r"[0-9a-f]{32}", r.headers["x-request-id"])


def test_replaces_unsafe_incoming_id():
    r = client.get("/hello", headers={"X-Request-ID": "bad id with spaces"})
    assert re.fullmatch(r"[0-9a-f]{32}", r.headers["x-request-id"])


def test_logs_carry_request_id():
    # capture_logs swaps the processor chain, so merge_contextvars must be passed back in.
    with capture_logs(processors=[merge_contextvars]) as logs:
        client.get("/hello", headers={"X-Request-ID": "req-1"})
    hello = [e for e in logs if e["event"] == "hello_called"]
    assert hello[0]["request_id"] == "req-1"


def test_context_cleared_after_error():
    with capture_logs(processors=[merge_contextvars]) as logs:
        r1 = client.get("/boom", headers={"X-Request-ID": "fail-1"})
        r2 = client.get("/hello")
    assert r1.status_code == 500
    hello = [e for e in logs if e["event"] == "hello_called"]
    assert hello[0]["request_id"] == r2.headers["x-request-id"] != "fail-1"

[pytest]
pythonpath = .