Using Python contextvars for Async Context Management

When you run concurrent code in Python, you often need to pass state that should be visible to a task and its children without threading it through every function signature. The python contextvars module provides a way to store such state in a context that is automatically propagated across async tasks and threads. This article explains how contextvars work, how to use them correctly, and where they fit compared to alternatives like thread-local storage or explicit parameters.

Why ContextVars Exist

Before contextvars, the common way to carry per-task state was threading.local(). That works when each thread runs a single logical operation, but it breaks down with asyncio. An event loop runs many tasks on the same thread, and tasks can be suspended and resumed at arbitrary points. If you store state in a thread-local, every task on that thread sees the same value, which is rarely what you want.

Contextvars solve this by associating state with a logical execution context, not a thread. When an asyncio task is created, it captures the current context. When the task runs, it sees the values from that context, even if it runs on the same thread as other tasks with different contexts.

The same mechanism works across threads. When you use loop.run_in_executor or asyncio.to_thread, the context is copied to the worker thread, so the code there sees the same context values as the caller.

The Core API: ContextVar, set, get, reset

The contextvars module provides ContextVar, Context, and the copy_context() function. The most common usage is creating a ContextVar and using its set and get methods.

import contextvars

user_id_var = contextvars.ContextVar("user_id", default=None)

def handle_request():
    user_id = user_id_var.get()
    print(f"Handling request for user {user_id}")

user_id_var.set(42)
handle_request()  # prints: Handling request for user 42

The set method returns a Token object. You can use that token to restore the previous value later with reset. This is important when you need to temporarily change a value and then revert it, especially in a try/finally block.

def set_user_id(user_id):
    token = user_id_var.set(user_id)
    try:
        # do work with the new value
        pass
    finally:
        user_id_var.reset(token)

Without resetting, the value would leak into the surrounding context, which can cause subtle bugs in long-running services.

How ContextVars Propagate Across Async Tasks

The key behavior is that asyncio tasks capture the current context at creation time. When you call asyncio.create_task(coro), the task runs with a copy of the context that was active at the moment create_task was called. This means that if you set a context variable before creating a task, the task sees that value. If you set it after the task is created, the task does not see the change.

import asyncio
import contextvars

request_id_var = contextvars.ContextVar("request_id", default=None)

async def worker():
    await asyncio.sleep(0.1)
    print(f"worker sees request_id: {request_id_var.get()}")

async def main():
    request_id_var.set("req-123")
    task = asyncio.create_task(worker())
    request_id_var.set("req-456")
    await task

asyncio.run(main())

This prints worker sees request_id: req-123 because the task captured the context before the second set. This behavior is intentional and lets you isolate state per task.

When you await a coroutine directly (without creating a task), the context is not copied; the coroutine runs in the same context as the caller. That means any set inside the coroutine is visible to the caller after the await returns, unless you explicitly reset it.

Running Code in a Separate Context

Sometimes you need to execute a function in a fresh context, or in a context that you control. The Context class and copy_context() function allow this.

contextvars.copy_context() returns a shallow copy of the current context. You can then modify it and run a function inside that copy using context.run(callable, *args, **kwargs). The function sees the values from the copied context, and any changes it makes to context variables are isolated to that copy.

import contextvars

var = contextvars.ContextVar("var", default="default")

def show():
    print(var.get())

ctx = contextvars.copy_context()
var.set("original")

ctx.run(show)  # prints: default (the copied context had no value set)

var.set("changed")
ctx.run(show)  # prints: default (the copy is still isolated)

You can also set values inside the copied context before running the function:

ctx = contextvars.copy_context()
ctx.run(var.set, "inside")
ctx.run(show)  # prints: inside

This pattern is useful when you want to run a block of code with a specific context without affecting the caller's context. It is also used internally by libraries that need to propagate context across execution boundaries, such as asyncio itself.

Common Mistakes and Edge Cases

One common mistake is assuming that context variables set inside a task are visible to the task's children. They are, because the child task captures the context at creation time, which includes the parent's current values. But if you set a variable inside a child task, it does not affect the parent's context.

Another pitfall is using contextvars with synchronous code that runs in a thread pool. When you submit a function to a thread pool via concurrent.futures.ThreadPoolExecutor, the context is not automatically propagated. The asyncio.to_thread function does propagate the context, but plain executor.submit does not. If you need context propagation in a thread pool, you must explicitly copy the context and run the function inside it.

import asyncio
import contextvars
from concurrent.futures import ThreadPoolExecutor

var = contextvars.ContextVar("var", default="default")

def thread_worker():
    print(var.get())

async def main():
    var.set("main")
    loop = asyncio.get_running_loop()
    # This does NOT propagate the context
    await loop.run_in_executor(None, thread_worker)  # prints: default

    # This propagates the context
    await asyncio.to_thread(thread_worker)  # prints: main

asyncio.run(main())

Also be careful with context variable mutation. If you store a mutable object as the value, changes to that object are visible across all contexts that share it. Contextvars do not deep-copy values; they only copy the binding between variable and value. If you need isolation, store immutable values or copy the object before storing.

Performance and Overhead

Contextvars add a small overhead to task creation and context switching. When a task is created, the current context is copied. The copy is shallow, so it only copies the mapping of variables to values, not the values themselves. For most applications, this overhead is negligible compared to the cost of I/O or other work.

However, if you create thousands of tasks per second, the context copying can become measurable. In such cases, consider whether you really need context propagation for every task, or whether you can set the context once at a higher level and reuse it. The copy_context() function is also relatively cheap, but using it excessively in tight loops can add up.

There is no built-in way to disable context propagation in asyncio. If you need to avoid the overhead for a specific task, you can run it in a separate context that you create manually, but that is rarely necessary.

When to Use ContextVars vs Other Approaches

Contextvars are the right tool when you need to pass state implicitly through a call chain, especially in async code. They are ideal for request-scoped data like user IDs, request IDs, tracing spans, or database session bindings.

If you only need to pass a value to a single function, an explicit parameter is simpler and more readable. Contextvars add indirection, so use them only when the value would otherwise be threaded through many layers or when you need to propagate across task boundaries.

Compared to thread-local storage, contextvars are the correct choice for asyncio code. Thread-local storage is still useful for code that runs exclusively in threads and does not mix with async. For example, a library that manages a connection per thread might use threading.local internally, but if that library is used in async code, it should switch to contextvars to avoid cross-task contamination.

A practical pattern is to create a context variable for each piece of request-scoped state and set it at the entry point of a request handler. Then all functions called during that request can access it without passing it explicitly. This keeps the code clean and avoids signature changes when new state is added.

import contextvars

current_user = contextvars.ContextVar("current_user", default=None)

def get_current_user():
    return current_user.get()

def process_request(user):
    token = current_user.set(user)
    try:
        # all downstream calls can use get_current_user()
        pass
    finally:
        current_user.reset(token)

Remember to reset the variable after the request is complete to prevent leakage into subsequent requests, especially in a long-running server where the same context might be reused.