Context Variables and Thread Safety in Python Logging

Concurrent Python services need request-scoped state - a trace ID, a tenant, a user - to reach every log line without being threaded through every function signature. The naive answer, threading.local(), quietly corrupts that state the moment a single OS thread multiplexes many coroutines. This guide explains why contextvars is the correct primitive, how its token-based lifecycle works, and how to propagate context safely across coroutines, thread pools, and logging handlers. It is part of the Python Logging Fundamentals and Structured Data reference and works hand in hand with structured logging with the standard library and the broader formatter configuration patterns.

Prerequisites

contextvars ships in the standard library from Python 3.7, so the core mechanics need no installation. The examples assume Python 3.11 or newer. For the framework integration sections, pin a web framework and, optionally, the OpenTelemetry API used to seed real trace identifiers:

pip install \
  "starlette>=0.37.0,<1.0.0" \
  "opentelemetry-api>=1.30.0,<2.0.0"

Set the service identity so emitted records carry a consistent resource label:

export OTEL_SERVICE_NAME="orders-api"

Concept and Architecture

A ContextVar is a named slot whose value is resolved against the current context rather than the current thread. The current context is an immutable mapping that the interpreter snapshots and restores automatically at coroutine switch points. That is the whole reason contextvars exists: under asyncio, one OS thread interleaves thousands of requests, so any state keyed by thread identity is shared across all of them. A value written by request A would be read by request B the instant the loop switched, producing trace ID collisions and cross-request data leaks.

Binding state to the execution context instead gives each logical flow its own isolated view. When asyncio.create_task() schedules a coroutine, it copies the current context, so the child starts with the parent's values but its own mutations stay private. This is structural isolation - no locks, no manual save-and-restore.

The lifecycle is token-based and deliberately explicit. ContextVar.set(value) returns a Token that records the previous state. Passing that token to ContextVar.reset(token) restores the prior value. The discipline matters: because each set() layers a new mapping, a request that sets but never resets leaves that layer in place, and in a long-lived worker those layers accumulate into unbounded growth. Always pair a set() with a reset() in a finally block, or wrap the pair in a context manager.

It is worth being precise about what "the current context" is. At any instant the interpreter holds an immutable Context mapping ContextVar objects to values. A get() is a lookup in that mapping; a set() does not mutate the mapping in place but installs a new value and hands back a token describing how to undo it. Because the mapping is immutable, snapshotting it is cheap and safe to share - that immutability is precisely what makes automatic copying across coroutine boundaries correct without any locking. Two tasks holding snapshots of the same context cannot corrupt each other's view of a scalar value, because neither can rewrite the other's mapping.

One boundary breaks the automatic copying: code that leaves the event loop. A concurrent.futures.ThreadPoolExecutor runs work on a separate thread that the loop did not create through create_task, so it does not inherit the caller's context. The fix is to snapshot the live context with contextvars.copy_context() and execute the work through Context.run(), which restores that snapshot inside the worker. The same explicit snapshot is what carries identifiers safely when you fan out across threads in thread-safe logging in multiprocessing, where the process boundary removes even the option of automatic copying.

The process boundary deserves emphasis because it is the most common misconception. A ContextVar lives inside one interpreter. When a ProcessPoolExecutor or a multiprocessing worker starts, the child either forks - inheriting a copy of memory frozen at fork time, not the value set later in the request - or spawns fresh with no inherited values at all. Either way, the running request's identifiers do not travel to the child automatically. To carry a trace ID across a process you must serialize it into the work item, the queue payload, or the network request, and re-establish the ContextVar inside the child before it logs. Treating cross-process propagation as a serialization problem rather than a context problem is the mental model that keeps it correct.

Step-by-Step Implementation

1. Declare the context variables at module scope. Define each ContextVar once, with a sentinel default, so every importer shares the same slot. A None or zero default lets a logging filter detect a missing value and substitute a fallback rather than raising.

import contextvars

request_id_ctx: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "request_id", default=None
)
trace_id_ctx: contextvars.ContextVar[str] = contextvars.ContextVar(
    "trace_id", default="0" * 32
)

2. Set the values at the request boundary and reset on teardown. Middleware is the natural place: it owns the full request lifecycle, so it can guarantee the matching reset. Capture the token in the same scope as the set.

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


class ContextMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        token = request_id_ctx.set(
            request.headers.get("X-Request-ID") or str(uuid.uuid4())
        )
        try:
            return await call_next(request)
        finally:
            # reset runs whether the handler succeeds or raises
            request_id_ctx.reset(token)

3. Read the context inside a logging filter, never on the handler. A handler instance is long-lived and shared, so caching a value on it would freeze the first request's identifiers onto every later record. Resolve inside filter() so each record reflects the context active at emission.

import logging


class ContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_ctx.get() or "no-request"
        record.trace_id = trace_id_ctx.get()
        return True

4. Propagate explicitly across the thread-pool boundary. Snapshot the caller's context and run the offloaded callable through it so the worker thread sees the same identifiers.

import concurrent.futures


def run_in_context(executor, fn, *args):
    ctx = contextvars.copy_context()  # snapshot the caller's context
    return executor.submit(ctx.run, fn, *args)

5. Keep the filter on the handler, not the logger, when handlers differ. Attaching the ContextFilter to a handler means every record that handler emits is enriched, regardless of which logger produced it. Attaching it to a logger enriches only that logger's records. For a uniform JSON contract across the whole service, put the filter on the shared handler so third-party libraries' log lines also carry the trace context.

import logging

handler = logging.StreamHandler()
handler.addFilter(ContextFilter())  # enriches every record this handler emits
logging.getLogger().addHandler(handler)

This separation - context lives in a ContextVar, enrichment lives in a Filter, serialization lives in a Formatter - is the design that keeps each piece independently testable and reusable across handler architectures, the same layering used throughout formatter configuration.

Configuration Reference

Async and Concurrency Considerations

The split that trips teams up is between coroutines and threads. asyncio.create_task(), asyncio.gather(), and asyncio.TaskGroup all copy the current context for the new task, so a trace ID set before the spawn is visible inside it automatically - and changes made inside the task do not leak back to the parent. That is exactly the isolation you want for per-request state.

A ThreadPoolExecutor is different. loop.run_in_executor() and executor.submit() hand work to a thread the loop did not create as a task, so no copy happens. The worker runs under whatever context that thread last held, which is usually the default. The copy_context() plus Context.run() pattern from step four is mandatory here; without it, offloaded CPU-bound work logs under the wrong trace ID or none at all. Snapshotting adds only a few microseconds per submission, negligible against the cost of the thread handoff itself.

Generators and context managers add a subtlety: a generator captures context when it is created, not when it resumes, so a ContextVar mutated between two next() calls may not be visible inside the generator body in the way you expect. For request tracing this rarely bites, but it is the reason long-lived background tasks should set their own context at startup rather than relying on inheriting one from whatever scheduled them. The end-to-end middleware-to-coroutine flow is worked through in using contextvars for request tracing.

Background schedulers warrant a deliberate convention. A periodic job, a consumer pulling from a broker, or a retry worker has no inbound request to inherit from, so it should mint a fresh trace ID at the start of each unit of work and reset it when that unit completes. Anchoring the set/reset pair to the boundary of one job - one message, one tick, one batch - keeps the context mapping flat and gives every emitted log line an identifier you can group on, even though no HTTP request was ever involved. The alternative, leaving the variable at its default, produces a stream of indistinguishable background log lines that are impossible to correlate during an incident.

There is also a measurable cost worth budgeting for. A get() is effectively free on the hot path, but set() allocates, so the rule of thumb is one set() per logical scope rather than per log line. In a request handler that means setting identifiers once in middleware and letting hundreds of downstream log calls read them, never re-setting per call. Snapshotting via copy_context() adds a few microseconds per thread-pool submission - immaterial next to the thread handoff itself, but a reason not to snapshot inside a tight loop. These costs are small enough that correctness, not performance, should drive the design; the failure mode that actually hurts is unbounded context growth from missing resets, not lookup latency.

Production Code Examples

End-to-end: filter injection with safe scoping

This module wires a ContextVar into a JSON-emitting logger and demonstrates the set-and-reset discipline a request handler must follow.

import contextvars
import json
import logging

request_id_ctx: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "request_id", default=None
)


class ContextFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.request_id = request_id_ctx.get() or "no-request"
        return True


class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        return json.dumps({
            "ts": self.formatTime(record),
            "level": record.levelname,
            "msg": record.getMessage(),
            "request_id": getattr(record, "request_id", None),
        })


logger = logging.getLogger("app")
logger.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(JSONFormatter())
_handler.addFilter(ContextFilter())
logger.addHandler(_handler)


def handle_request(req_id: str) -> None:
    token = request_id_ctx.set(req_id)
    try:
        logger.info("processing request")
    finally:
        request_id_ctx.reset(token)


if __name__ == "__main__":
    handle_request("req-8842")

Expected Output:

{"ts": "2026-06-19 12:00:00,000", "level": "INFO", "msg": "processing request", "request_id": "req-8842"}

Explicit propagation into a thread pool

This shows the snapshot-and-run pattern that keeps the trace identifier intact when work is offloaded to worker threads.

import concurrent.futures
import contextvars

trace_id_ctx: contextvars.ContextVar[str] = contextvars.ContextVar(
    "trace_id", default="root"
)


def _process(item_id: str) -> None:
    # Runs on a worker thread but reads the caller's trace ID.
    print(f"processing {item_id} under {trace_id_ctx.get()}")


def main() -> None:
    token = trace_id_ctx.set("trace-abc-123")
    try:
        ctx = contextvars.copy_context()  # capture the active context once
        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
            futures = [
                pool.submit(ctx.run, _process, f"task-{i}") for i in range(3)
            ]
            concurrent.futures.wait(futures)
    finally:
        trace_id_ctx.reset(token)


if __name__ == "__main__":
    main()

Expected Output:

processing task-0 under trace-abc-123
processing task-1 under trace-abc-123
processing task-2 under trace-abc-123

End-to-end: a worker-pool service

The pattern that exercises every boundary at once is an async service that offloads CPU-bound work to a thread pool while keeping logs correlated. The middleware sets the trace ID, async handlers inherit it for free, and the offloaded work re-establishes it explicitly.

import asyncio
import concurrent.futures
import contextvars
import json
import logging

trace_id_ctx: contextvars.ContextVar[str] = contextvars.ContextVar(
    "trace_id", default="no-trace"
)
_pool = concurrent.futures.ThreadPoolExecutor(max_workers=4)


class TraceFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.trace_id = trace_id_ctx.get()
        return True


class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        return json.dumps({
            "level": record.levelname,
            "msg": record.getMessage(),
            "trace_id": getattr(record, "trace_id", "no-trace"),
        })


log = logging.getLogger("svc")
log.setLevel(logging.INFO)
_h = logging.StreamHandler()
_h.setFormatter(JSONFormatter())
_h.addFilter(TraceFilter())
log.addHandler(_h)


def cpu_work(n: int) -> int:
    # Runs on a worker thread; reads the trace ID restored by ctx.run.
    log.info(f"hashing batch {n}")
    return n * n


async def handle(trace_id: str) -> int:
    token = trace_id_ctx.set(trace_id)
    try:
        log.info("request received")  # inherits trace_id directly
        loop = asyncio.get_running_loop()
        ctx = contextvars.copy_context()  # snapshot before crossing to a thread
        result = await loop.run_in_executor(_pool, lambda: ctx.run(cpu_work, 7))
        log.info("request done")
        return result
    finally:
        trace_id_ctx.reset(token)


async def main() -> None:
    await asyncio.gather(handle("trace-aaa"), handle("trace-bbb"))


if __name__ == "__main__":
    asyncio.run(main())
    _pool.shutdown()

Expected Output:

{"level": "INFO", "msg": "request received", "trace_id": "trace-aaa"}
{"level": "INFO", "msg": "request received", "trace_id": "trace-bbb"}
{"level": "INFO", "msg": "hashing batch 7", "trace_id": "trace-aaa"}
{"level": "INFO", "msg": "hashing batch 7", "trace_id": "trace-bbb"}
{"level": "INFO", "msg": "request done", "trace_id": "trace-aaa"}
{"level": "INFO", "msg": "request done", "trace_id": "trace-bbb"}

The two concurrent requests never cross-contaminate: each handle coroutine carries its own isolated trace_id, the async log lines inherit it automatically, and the thread-pool log lines carry it only because ctx.run restored the snapshot. Drop the copy_context() and the hashing batch lines fall back to no-trace, which is the exact symptom that flags a missing boundary snapshot in production.

Common Mistakes

Mutating a ContextVar without capturing the token. Calling var.set(value) and discarding the return value leaves no way to restore the previous state. The error surfaces later as RuntimeError: cannot reset context: token created in a different Context, or worse, as silent trace ID bleed between requests. Capture the token in the same scope as the set and pass it to reset in a finally block.

Assuming a thread pool inherits async context. A coroutine that calls loop.run_in_executor() and expects the worker to see its trace ID sees the default instead, because copying only happens for tasks created on the event loop. The symptom is log lines from offloaded work tagged root or no-request. Wrap the call with copy_context() and Context.run().

Caching a context value on a long-lived object. Reading a ContextVar once in a handler's __init__ and reusing it freezes the first request's identifiers onto every subsequent record. The symptom is every log line sharing one trace ID. Resolve the value inside the per-record filter() method so it always reflects the active context.

Storing mutable objects in a ContextVar and sharing the reference. Two coroutines that both hold the same dict from a context variable can race on its contents even though the slot itself is isolated. Store immutable primitives, or take a defensive copy when seeding the context, so isolation extends to the value as well as the slot.

Re-setting the variable on every log call. Some teams set the trace ID immediately before each logger.info out of caution, which both wastes the set() allocation and risks leaving the context one layer deeper than the matching reset count. Set identifiers once at the scope boundary and let the filter resolve them per record.

Frequently Asked Questions