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 the concerns listed above, 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 web frameworks such as FastAPI and Starlette 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 every Context is backed by a HAMT (Hash Array Mapped Trie33 Bagwell, Phil. Ideal Hash Trees (2001), the paper that introduced HAMTs. Clojure’s persistent maps and Scala’s immutable HashMap are the most widely deployed implementations. ), a persistent data structure: setting a variable does not mutate the existing trie but produces a new root that shares all untouched branches with it. A copy of a context is therefore just a pointer to its current root, which is why copy_context() is cheap and why each ContextVar.set inside a task is invisible to every other task. The per-task lookup is a single pointer dereference plus a trie walk. ⊕A reader familiar with Rust will recognize this pattern as Tokio’s task_local!: each task holds its own value, shadows nothing across tasks. The read path is lock-free because the data is per-thread and the structure itself is immutable.

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 is to bind one ContextVar per key, or wrap the whole bundle in a frozen dataclass and store one ContextVar of that bundle.

Sharing context in an async runtime

A ContextVar binding lives on the task that set it, and only on that task. Two boundaries break that assumption, and in both cases the value is set but code downstream never sees it.

The first boundary is sync-async. Context travels with the task, not with the OS thread, so any hop from a task onto a bare thread worker loses the binding. A def endpoint dependency runs in a Starlette thread-pool worker, so calling bind_contextvars(request_id=...) there never reaches the async endpoint.

The structlog docs state it directly: “context variables set in a synchronous context don’t appear in logs from an async context and vice versa.” The fix is to bind from code that runs inside the request task: an async middleware (the example above) or BaseHTTPMiddleware.dispatch. 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 boundary is the process boundary. multiprocessing and concurrent.futures.ProcessPoolExecutor cannot ship a Context across processes; PEP 567 explicitly rejects making context objects picklable. The workaround is to extract the values into a plain dict in the parent and re-bind in the child after fork:

# 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])

Fork copies the parent’s bindings once, at the moment of fork; any ContextVar.set in the child afterwards is invisible to the parent and to other workers. 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 (gunicorn design docs).

The operational rule is the same for both boundaries: bind cross-cutting state from a per-request middleware, never from startup code, where the binding lives on the loop’s main task and is shared by every subsequent request.

What about threading.local?

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, the closest equivalent to ContextVar is Tokio’s task_local!. Both give each task its own private binding: a value set inside one task is invisible to its siblings, and the value the task reads is the one bound on its own chain. The differences are mechanical. task_local! scopes its value to the future it wraps, so the value exists only inside that future’s body; a ContextVar can be set anywhere and read from any task that inherited the binding through await boundaries, which is what makes middleware-set values visible deep in the call stack. task_local! also requires an explicit scope(...) call to enter the value’s lifetime, while the context machinery in Python propagates implicitly through task creation.

ContextVar is for ambient context the task inherits (request ID, locale, trace ID), and task_local! matches that use exactly. For shared state that every task reads and writes (a counter, a connection pool), neither primitive fits; use a properly synchronized container or a pool handle injected at the boundary.

References and additional resources

• Python standard library: typing, contextvars