Cross-Cutting Concerns Without Constructor Over-Injection

Using ContextVar for context propagation

Constructor injection makes sense for dependencies the service uses (repositories, gateways, in-process caches). For cross-cutting concerns that almost every service needs (a request ID on every log line, a trace ID propagated to every outbound call, the authenticated user’s identity, the current locale), constructors’ parameter list grows longer and longer.

The Python answer is contextvars.ContextVar11 PEP 567 (Python 3.7+) introduced contextvars for per-task context propagation across await boundaries. structlog, opentelemetry, and modern web frameworks use it for request-scoped metadata. . Each asyncio task gets its own copy of the context, and ContextVar.set / ContextVar.get propagates correctly across await boundaries:

from contextvars import ContextVar

request_id_var: ContextVar[str] = ContextVar("request_id", default="-")
current_user_var: ContextVar[str] = ContextVar("current_user", default="anonymous")

A FastAPI middleware sets the values per request:

from uuid import uuid4
from fastapi import Request
from shop.observability import current_user_var, request_id_var

async def request_context_middleware(request: Request, call_next):
    token_rid = request_id_var.set(request.headers.get("x-request-id", str(uuid4())))
    token_user = current_user_var.set(request.headers.get("x-user", "anonymous"))
    try:
        return await call_next(request)
    finally:
        request_id_var.reset(token_rid)
        current_user_var.reset(token_user)

A service reads the values without any constructor parameter:

import structlog
from shop.observability import request_id_var

logger = structlog.get_logger()

class OrderProcessorService:
    def __init__(self, repository, payment_gateway):
        self._repository = repository
        self._payment_gateway = payment_gateway

    async def execute(self, order_id):
        logger.info("processing_order", order_id=order_id, request_id=request_id_var.get())
        # remaining body invokes self._payment_gateway and self._repository the same way

The test sets the context explicitly, with the same try/finally pattern we saw in the DI post:

async def test_log_includes_request_id():
    token = request_id_var.set("test-rid-1")
    try:
        # invoke service.execute(order_id=42); assert logs contain order_id=42 and request_id='test-rid-1'
        ...
    finally:
        request_id_var.reset(token)

Why per-task context is safe under concurrency

The middleware example above works because asyncio.Task snapshots the running context at construction time. From cpython/Lib/asyncio/tasks.py:

class Task(futures._PyFuture):
    def __init__(self, coro, *, loop=None, name=None, context=None, ...):
        ...
        if context is None:
            self._context = contextvars.copy_context()
        else:
            self._context = context

Every await inside the task resumes inside that snapshot. Two concurrent FastAPI requests each run on their own task, and each task received an independent copy_context() snapshot at the moment asyncio.create_task was called. The middleware’s set call mutates only the copy that lives for that task’s lifetime, which is the same reason request B’s request_id cannot appear in request A’s logs.

The same isolation holds across OS threads. Each thread has its own ts->context pointer inside the interpreter, and writes in one thread are invisible to reads in another. The cpython test suite proves this in ContextTest.test_context_threads_1: ten threads each set the same ContextVar to a unique value across a hundred iterations, and every thread’s get() returns the value it just set. The contextvars docs state the rule normatively: “Since each thread has its own context stack, ContextVar objects behave in a similar fashion to threading.local() when values are assigned in different threads.”

The one place this guarantee breaks is loop.run_in_executor. The worker thread does not inherit the calling task’s context, because BaseEventLoop.run_in_executor calls executor.submit(func, *args) without a context parameter. The calling coroutine’s ContextVar values are not visible inside the offloaded function. The fix is the idiom PEP 567 prescribes in its examples section:

from concurrent.futures import ThreadPoolExecutor
import contextvars

executor = ThreadPoolExecutor()
current = contextvars.copy_context()
executor.submit(current.run, some_function)

contextvars.copy_context() snapshots the calling task’s bindings; Context.run(func) replays them on the worker thread for the duration of the call. There is no global lock, because the HAMT that backs every Context is a persistent (immutable) tree: each ContextVar.set returns a new context root, and the per-task lookup is a single pointer dereference plus a HAMT search. A reader familiar with Rust will recognize the shape: ContextVar is the per-task equivalent of an Rc<RefCell<>> that is automatically swapped on every Task boundary, and the read path is lock-free because the data is per-thread.

How structlog and opentelemetry use ContextVar

The hynek/structlog contextvars.py module uses a dict of ContextVars, not a single ContextVar[dict]. The author’s own comment explains the reason: a single ContextVar holding a dict would let any code that reads the dict see the current task’s values, not the values bound when the log line was emitted. One ContextVar per key gives proper per-task isolation.

The read path, merge_contextvars, iterates contextvars.copy_context() on every logger.info(...) call and copies each structlog_-prefixed binding into the event dict with setdefault, so a single bind_contextvars(request_id=...) at request entry propagates to every downstream log line.

The opentelemetry-python contextvars_context.py module takes the opposite choice: one ContextVar whose value is the OTel Context dict. It works because OTel’s Context is treated as immutable: callers call set_value to obtain a new Context, then pass it to attach(token), which is ContextVar.set with one indirection:

class ContextVarsRuntimeContext(_RuntimeContext):
    def __init__(self) -> None:
        self._current_context = ContextVar(self._CONTEXT_KEY, default=Context())

    def attach(self, context: Context) -> Token[Context]:
        return self._current_context.set(context)

    def detach(self, token: Token[Context]) -> None:
        self._current_context.reset(token)

The two designs are not interchangeable. structlog needs per-key isolation because a log line might be emitted while another task is binding the same key; OTel’s Context only carries the current trace metadata, so a single ContextVar with a value object suffices. The takeaway for application code: if you have more than one piece of context to bind, do not store them in a single ContextVar[dict]. Use one ContextVar per key, or wrap the whole bundle in a frozen dataclass and store one ContextVar of that bundle.

Where this breaks

Three production-time failure modes are worth knowing up front. The structlog docs name the first directly: “context variables set in a synchronous context don’t appear in logs from an async context and vice versa.” This is a Starlette/FastAPI-specific gotcha. A def endpoint dependency that calls bind_contextvars(request_id=...) runs in a thread-pool worker that does not inherit the request task’s context, so the binding never reaches the async endpoint.

The fix is to bind from an async middleware (the example above) or from BaseHTTPMiddleware.dispatch, which runs inside the request task. The full thread is fastapi/fastapi#5999.22 From the structlog docs (Context Variables): the framework warns that “context variables set in a synchronous context don’t appear in logs from an async context and vice versa.” Confirmed in practice by fastapi/fastapi#5999, where Starlette maintainer Kludex notes that the binding must be set inside an async path to propagate to the async endpoint.

The second mode is process-boundary loss. multiprocessing and concurrent.futures.ProcessPoolExecutor cannot ship a Context across the process boundary. The workaround is to extract the values into a plain dict in the parent and re-bind in the child after fork. PEP 567 lists this under “Rejected Ideas”:

# parent
values = {name: var.get() for name, var in context_vars.items()}

# child, after fork
for name, var in context_vars.items():
    var.set(values[name])

A worker process forked from a parent that bound request_id_var inherits the binding at the moment of fork, but any subsequent ContextVar.set in the child is independent of the parent. The gunicorn preload_app=True case is the canonical version: the app runs in the parent, forks workers, and each worker either rebinds per-request in its own middleware or carries stale bindings until the next request. Bind cross-cutting state from a per-request middleware, not from app.on_startup, where the binding lives on the loop’s main task and is shared by every subsequent request.

Compared to the alternatives

threading.local() is the wrong primitive for async code. Every asyncio task scheduled on one event loop runs on the same OS thread, so threading.local writes are visible across coroutines, and asyncio.gather will mix their values. The contextvars docs recommend ContextVar over threading.local for any state that crosses concurrent code.

Coming from Rust, one might think ContextVar is similar to Rc<Mutex<...>>, but they are serving conceptually different use cases. Rc<Mutex<...>> is a shared, mutable container protected by a runtime lock; every read contends, every write requires a lock, and the global map is a single point of contention across all tasks. ContextVar is the opposite: each task holds its own immutable snapshot, the read path is a pointer dereference, and no lock is required. The closest equivalent in Rust would be Tokio’s task_local!.

The two primitives solve different problems. ContextVar is for ambient context the task inherits (request ID, locale, trace ID); Rc<Mutex<HashMap>> is for shared state every task reads and writes (a counter, a connection pool).

References and additional resources

• Python standard library: typing, contextvars
• PEP 567 (contextvars): https://peps.python.org/pep-0567/