Context Variables

Warnings

• gotcha Creating `ContextVar` instances within closures or functions repeatedly (e.g., inside a loop or request handler) can lead to memory leaks in long-running applications. Each call creates a new `ContextVar` object that may persist in the context graph.
  Fix: Always declare `ContextVar` instances at the module level or as class attributes to ensure a single instance is created per application lifetime.
• gotcha Calling `ContextVar.reset(token)` with a token that was created in a different `Context` will raise a `ValueError`. Additionally, a `RuntimeError` is raised if a token is used to reset a variable more than once.
  Fix: Ensure `reset(token)` is called in the same `Context` where its corresponding `set()` operation occurred and that each token is used only once. Using `try-finally` blocks or `contextlib.contextmanager` is a common pattern to manage `set`/`reset` pairs safely.
• gotcha Forgetting to `reset()` a `ContextVar` after `set()` in long-running applications (e.g., web servers, background workers) can lead to state bleeding, where subsequent tasks inherit an unexpected context value. This is a common source of hard-to-debug issues.
  Fix: Always pair `ContextVar.set()` with `ContextVar.reset(token)` in a `try-finally` block to guarantee the context is restored, even if exceptions occur. Python 3.14+ `Token` objects support the context manager protocol for automatic reset.
• gotcha Attempting to get the value of a `ContextVar` using `ContextVar.get()` when no value has been set in the current context and no `default` value was provided during `ContextVar` instantiation will raise a `LookupError`.
  Fix: Always provide a `default` value when creating a `ContextVar` if a value isn't guaranteed to be set, or wrap `ContextVar.get()` calls in `try-except LookupError` blocks. Alternatively, `ContextVar.get(default_value_for_this_call)` can be used.
• gotcha Contexts are not automatically propagated across process boundaries. If `contextvars` are used with `multiprocessing.ProcessPoolExecutor`, changes made in a worker process's context will not affect the main process's context or other worker processes, as context variables are copied (pickled) when tasks are assigned.
  Fix: Avoid using `contextvars` for sharing state directly between `ProcessPoolExecutor` processes. Instead, use multiprocessing primitives (e.g., `Queue`, `Pipe`, `shared_memory`) for inter-process communication if state needs to be truly shared or synchronized.

Install

• pip install contextvars Install `contextvars`

Imports

• ContextVar

  from contextvars import ContextVar

• copy_context

  from contextvars import copy_context

• current_context

  from contextvars import current_context

  Used to retrieve the current Context object as a mapping.
• Token

  from contextvars import Token

  An opaque object returned by `ContextVar.set()` and required by `ContextVar.reset()` to revert a variable to its previous state.

Quickstart

This example demonstrates how `ContextVar` isolates state in asynchronous tasks. Each `handle_request` coroutine sets a `current_user` value, which remains local to its execution context, preventing state bleeding between concurrent tasks. The `token` returned by `set()` is crucial for `reset()`ing the variable to its previous state, often done in a `try-finally` block for proper cleanup.

import asyncio
import contextvars

# Define a context variable
current_user = contextvars.ContextVar('current_user', default='Guest')

async def handle_request(user_name):
    # Set the user for the current task. This returns a Token.
    token = current_user.set(user_name)
    try:
        print(f"Task for {user_name}: Currently handled by {current_user.get()}")
        await asyncio.sleep(0.1) # Simulate async work
        print(f"Task for {user_name}: Still handled by {current_user.get()}")
    finally:
        # Reset the context variable to its previous value using the token
        current_user.reset(token)

async def main():
    print(f"Before tasks: {current_user.get()}")
    await asyncio.gather(
        handle_request("Alice"),
        handle_request("Bob")
    )
    print(f"After tasks: {current_user.get()}") # Should revert to default

if __name__ == '__main__':
    asyncio.run(main())