Context propagation in FastAPI: request_id, trace_id and user_id in every log

← Back to the section

Imagine: you have 20 requests running at the same time, and in the logs the lines from all of them are mixed together. Which line belongs to which request? Without extra information there's no way to tell.

The solution is context propagation: set the request identifier (request_id) once at the entry point, and it automatically appears in every log record inside that request — in services, dependencies, background tasks. You don't need to drag it as a parameter through all your functions.

How it works in Python

In Java, logs are bound to a thread through MDC (thread-local storage). In Python with asyncio a single thread serves thousands of requests through coroutines. The mechanism is different — contextvars.ContextVar.

ContextVar stores a value separately for each asyncio task. When you create a task through await or asyncio.create_task, the child task automatically gets a copy of the parent's context. That means the request_id set at the start of a request is visible in all await calls inside that request without being passed as an argument.

The structlog library uses contextvars under the hood — the bind_contextvars / clear_contextvars functions work with the current asyncio task.

Middleware: set request_id once

The right place to set request_id is middleware. It intercepts every incoming request before it reaches the handler.

# app/middleware/request_id.py
import uuid
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from structlog.contextvars import bind_contextvars, clear_contextvars

class RequestIdMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = request.headers.get("X-Request-Id") or str(uuid.uuid4())
        bind_contextvars(request_id=request_id)
        try:
            response = await call_next(request)
            response.headers["X-Request-Id"] = request_id
            return response
        finally:
            clear_contextvars()

# app/main.py
from fastapi import FastAPI
from app.middleware.request_id import RequestIdMiddleware

app = FastAPI()
app.add_middleware(RequestIdMiddleware)

What happens here:

We read X-Request-Id from the request header — if the client already sent it (for example, from an API gateway), we keep it. If not, we generate a UUID.
bind_contextvars(request_id=request_id) writes the value into the contextvars of the current task.
We return the same request_id in the response header — the client can pass it to support if there are problems.
clear_contextvars() in the finally block is mandatory. Without it, when event-loop tasks are reused, the request_id from the previous request leaks into the logs of the next one.

Important ordering: the request_id middleware must be added before the auth middleware. Then even authentication errors will carry request_id in the logs.

trace_id and span_id: added by OpenTelemetry automatically

request_id helps you find all the logs of one request inside a single service. trace_id is an identifier that links logs and traces across several services. You don't need to set it manually — a structlog processor does it by reading the active OpenTelemetry span.

# app/observability/logging.py
import structlog
from opentelemetry import trace

def otel_trace_context_processor(logger, method, event_dict):
    span = trace.get_current_span()
    ctx = span.get_span_context()
    if ctx.is_valid:
        event_dict["trace_id"] = format(ctx.trace_id, "032x")
        event_dict["span_id"] = format(ctx.span_id, "016x")
    return event_dict

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,   # adds request_id, user_id
        otel_trace_context_processor,               # adds trace_id, span_id
        structlog.processors.JSONRenderer(),
    ]
)

merge_contextvars first pours in everything stored in contextvars (request_id, user_id), then the processor adds trace_id and span_id from the active OTel span. As a result every log record automatically contains all four fields.

Don't add trace_id through bind_contextvars by hand — the span changes during request processing, and a manually set value would go stale.

user_id: added after the token is verified

user_id can't be added in the request_id middleware — at that moment the token isn't verified yet. The right place is a dependency (Depends) that validates the JWT.

# app/dependencies/auth.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from structlog.contextvars import bind_contextvars

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> str:
    payload = decode_jwt(credentials.credentials)
    user_id = payload.get("sub")
    if not user_id:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    bind_contextvars(user_id=user_id)
    return user_id

# app/routers/orders.py
from fastapi import APIRouter, Depends
import structlog
from app.dependencies.auth import get_current_user

router = APIRouter()
log = structlog.get_logger(__name__)

@router.post("/orders")
async def create_order(
    payload: CreateOrderRequest,
    user_id: str = Depends(get_current_user),
):
    log.info("order_create_requested", customer_id=payload.customer_id)
    # request_id, user_id, trace_id will land in the record automatically

bind_contextvars(user_id=...) runs inside the request's asyncio task, and the value is available to all child coroutines. clear_contextvars() in the finally of the middleware will also drop user_id at the end of the request.

You must not call bind_contextvars inside a service or handler directly — there is no clear there, and where to call it becomes non-obvious.

asyncio.create_task: context is passed automatically

If inside a handler you start a background task through asyncio.create_task, the context is passed into it automatically — no extra code is needed.

# app/services/order_service.py
import asyncio
import structlog

log = structlog.get_logger(__name__)

async def confirm_order(order_id: str, customer_id: str) -> None:
    log.info("confirm_order_started", order_id=order_id)

    asyncio.create_task(
        _send_confirmation_email(order_id, customer_id)
    )

async def _send_confirmation_email(order_id: str, customer_id: str) -> None:
    log.info("email_queued", order_id=order_id, customer_id=customer_id)
    # request_id is already here — the Task got a copy of the context automatically

This is the key difference from Java: you don't need any analogue of TaskDecorator — asyncio copies contextvars when the task is created.

run_in_executor: context must be passed explicitly

Here's the one place where automatic propagation breaks. If you offload blocking code to a thread pool through run_in_executor, the threads don't get contextvars automatically. You need to take a snapshot of the context in advance.

# app/services/product_service.py
import asyncio
import contextvars
import structlog
from opentelemetry import context as otel_context

log = structlog.get_logger(__name__)

async def export_product_catalog(product_ids: list[str]) -> bytes:
    loop = asyncio.get_running_loop()
    ctx = contextvars.copy_context()       # snapshot of contextvars before the offload
    otel_ctx = otel_context.get_current()  # snapshot of the OTel context

    def _blocking_export():
        token = otel_context.attach(otel_ctx)
        try:
            return ctx.run(_build_csv, product_ids)
        finally:
            otel_context.detach(token)

    return await loop.run_in_executor(None, _blocking_export)

def _build_csv(product_ids: list[str]) -> bytes:
    log.info("catalog_export_started", count=len(product_ids))
    # request_id is available here through ctx.run
    ...

ctx.run(fn, *args) runs the function with the restored contextvars snapshot. otel_context.attach/detach restores the active OTel span — without it, new spans inside the thread would have no parent and wouldn't link to the trace.

Kafka and background queues: pass traceparent in headers

When sending an event to Kafka, the OTel context doesn't propagate on its own — you need to serialize it into the message headers.

# app/adapters/kafka/producer.py
from opentelemetry import propagate
from structlog.contextvars import get_contextvars
import structlog

log = structlog.get_logger(__name__)

async def publish_order_placed(order_id: str, customer_id: str) -> None:
    carrier: dict[str, str] = {}
    propagate.inject(carrier)  # traceparent + tracestate → carrier

    ctx_vars = get_contextvars()
    headers = {
        "traceparent": carrier.get("traceparent", ""),
        "request_id": ctx_vars.get("request_id", ""),
    }

    await kafka_producer.send(
        "order.placed",
        value={"order_id": order_id, "customer_id": customer_id},
        headers=list(headers.items()),
    )
    log.info("order_placed_published", order_id=order_id)

In the consumer we restore the context before processing:

# app/consumers/order_placed_consumer.py
from opentelemetry import propagate, context as otel_context
from structlog.contextvars import bind_contextvars, clear_contextvars

async def handle_order_placed(message) -> None:
    headers = dict(message.headers)
    carrier = {"traceparent": headers.get("traceparent", "")}
    parent_ctx = propagate.extract(carrier)

    bind_contextvars(
        request_id=headers.get("request_id", ""),
        order_id=message.value["order_id"],
    )
    token = otel_context.attach(parent_ctx)
    try:
        await _process_order_placed(message.value)
    finally:
        otel_context.detach(token)
        clear_contextvars()

propagate.extract restores the OTel context from the traceparent header — the consumer's span becomes a child of the producer's span. Without it, Tempo would show two unlinked trace trees instead of one end-to-end trace.

Common mistakes

A forgotten clear_contextvars() — the most dangerous mistake. If you don't clear the context in finally, when event-loop tasks are reused the user_id from one request lands in the logs of another user. That's a data leak.

bind_contextvars in a service — calling bind_contextvars deep in the code (not in middleware and not in Depends) means losing control over clear. Keep context setup in one place — at the entry point.

run_in_executor without copy_context() — the thread will have an empty context, request_id will disappear from the logs, spans will lose their parent.

bind_contextvars(trace_id=...) by hand — setting trace_id manually is pointless: the span changes during request processing, and the manually set value will be stale. Leave it to the OTel processor.

In short

contextvars.ContextVar is isolated storage per asyncio task. An analogue of MDC, but it works with asyncio without TaskDecorator.
Set request_id once in middleware through bind_contextvars — it will automatically land in all the logs of the request.
clear_contextvars() in the finally block of the middleware is mandatory, otherwise the context leaks into the next request.
Add user_id in the auth dependency (Depends), after the JWT is verified — not in the request_id middleware.
trace_id and span_id are added by the OTel–structlog processor automatically — don't set them by hand.
asyncio.create_task copies contextvars into the child task automatically.
run_in_executor doesn't inherit the context — before the offload take a snapshot through copy_context() and otel_context.get_current().
Kafka and queues: serialize traceparent into the message headers through propagate.inject, restore it in the consumer through propagate.extract.