Context Variables with contextvars

When you’re writing async code, you often need to track information that’s specific to the current execution context—like a request ID in a web server, user authentication data, or tracing information. The contextvars module provides a clean way to manage this kind of contextual state without worrying about threads or async tasks interfering with each other.

This guide covers ContextVar, how it differs from thread-local storage, and practical patterns for using it in async applications.

The Problem: Context in Async Code

Imagine you’re building a web API with asyncio. Each incoming request might trigger multiple async operations—database queries, external API calls, background tasks. You want to track a request ID through all of these:

import asyncio

# This doesn't work as expected
request_id = None

async def handle_request(req_id):
    global request_id
    request_id = req_id
    await fetch_user_data()
    await process_payment()
    print(f"Request {request_id} completed")

async def fetch_user_data():
    # Which request is this?
    print(f"Fetching data for {request_id}")
    await asyncio.sleep(0.1)

async def process_payment():
    print(f"Processing payment for {request_id}")
    await asyncio.sleep(0.1)

async def main():
    await handle_request("req-123")
    await handle_request("req-456")

asyncio.run(main())

The global variable gets overwritten when tasks interleave, producing incorrect output. This is where contextvars helps.

Introducing ContextVar

ContextVar creates variables that are local to the current async context:

from contextvars import ContextVar

request_id = ContextVar("request_id", default=None)

Each async task gets its own copy of the variable:

import asyncio
from contextvars import ContextVar

request_id = ContextVar("request_id", default="none")

async def fetch_data():
    # Each task sees its own value
    print(f"Fetching data for {request_id.get()}")
    await asyncio.sleep(0.1)

async def handle_request(req_id):
    # Set the value for this context
    token = request_id.set(req_id)
    try:
        await fetch_data()
        await fetch_data()
    finally:
        # Restore the previous value
        request_id.reset(token)

async def main():
    # Each request runs in its own context
    await asyncio.gather(
        handle_request("req-123"),
        handle_request("req-456"),
    )

asyncio.run(main())
# Output:
# Fetching data for req-123
# Fetching data for req-456

Now each request maintains its own request_id without interfering with others.

Setting and Getting Values

ContextVar provides three key methods:

• get() — Returns the current value (or default if not set)
• set(value) — Sets a new value, returns a token for reset
• reset(token) — Reverts to the value before the corresponding set()

from contextvars import ContextVar

user_context = ContextVar("user_context", default={"guest": True})

# Get the default
print(user_context.get())  # {'guest': True}

# Set a new value
token = user_context.set({"user": "alice", "guest": False})
print(user_context.get())  # {'user': 'alice', 'guest': False}

# Reset back to default
user_context.reset(token)
print(user_context.get())  # {'guest': True}

The token-based reset is useful for temporary overrides:

async def with_admin_privileges():
    token = user_context.set({"user": "admin", "role": "admin"})
    try:
        await do_admin_stuff()
    finally:
        user_context.reset(token)

Using copy_context()

When you want to run code in a new context that inherits values from the current one, use copy_context():

from contextvars import copy_context, ContextVar

trace_id = ContextVar("trace_id", default="")

async def child_task():
    # This runs in a copy of the parent's context
    print(f"Child sees trace_id: {trace_id.get()}")
    await asyncio.sleep(0.1)

async def main():
    trace_id.set("abc-123")
    
    # Create a copy of the current context
    ctx = copy_context()
    
    # Run child tasks in that context
    await asyncio.gather(
        child_task(),
    )

asyncio.run(main())

This is useful when spawning background tasks that should inherit the parent’s context.

ContextVars with asyncio

The asyncio module automatically manages context for tasks. When you create a new task with asyncio.create_task(), it gets a fresh context that inherits from the parent:

import asyncio
from contextvars import ContextVar

request_id = ContextVar("request_id", default=None)

async def log_request():
    # This runs in the task's context
    print(f"Logging request: {request_id.get()}")

async def handle_request(req_id):
    request_id.set(req_id)
    # Creating a task inherits the context
    task = asyncio.create_task(log_request())
    await task

async def main():
    await asyncio.gather(
        handle_request("req-1"),
        handle_request("req-2"),
    )

asyncio.run(main())
# Output:
# Logging request: req-1
# Logging request: req-2

Thread Safety

ContextVar works correctly across threads. Each thread gets its own context:

import threading
from contextvars import ContextVar
import asyncio

thread_local_var = ContextVar("thread_local_var", default="default")

def thread_worker():
    print(f"Thread {threading.current_thread().name}: {thread_local_var.get()}")

async def async_worker():
    thread_local_var.set("async value")
    print(f"Async task: {thread_local_var.get()}")

async def main():
    # Run in async context
    await async_worker()
    
    # Run in thread
    thread = threading.Thread(target=thread_worker)
    thread.start()
    thread.join()

asyncio.run(main())
# Output:
# Async task: async value
# Thread Thread-1: default

The async and thread contexts are separate.

Practical Patterns

Request ID Middleware

A common pattern in web frameworks:

from contextvars import ContextVar
import asyncio

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

async def middleware_handler(request):
    import uuid
    req_id = str(uuid.uuid4())[:8]
    token = request_id_var.set(req_id)
    try:
        return await route_handler(request)
    finally:
        request_id_var.reset(token)

async def route_handler(request):
    # Access request ID anywhere in the call chain
    req_id = request_id_var.get()
    await process_request(request)
    return {"request_id": req_id}

async def process_request(request):
    # No need to pass request_id through every function
    print(f"Processing {request_id_var.get()}")

Authentication Context

Store user information for the current request:

from contextvars import ContextVar

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

async def get_current_user():
    return current_user.get()

async def require_auth(handler):
    async def wrapper(request):
        user = await authenticate(request)
        token = current_user.set(user)
        try:
            return await handler(request)
        finally:
            current_user.reset(token)
    return wrapper

Tracing and Logging

Add consistent tracing across async operations:

from contextvars import ContextVar
import logging

trace_id = ContextVar("trace_id", default="no-trace")
logger = logging.getLogger(__name__)

class ContextFilter(logging.Filter):
    def filter(self, record):
        record.trace_id = trace_id.get()
        return True

# In your async code
async def handle_request():
    import uuid
    trace_id.set(str(uuid.uuid4()))
    logger.info("Starting request")
    await do_work()
    logger.info("Request complete")

ContextVar vs threading.local

Before contextvars, you might have used threading.local():

import threading

data = threading.local()
data.value = "hello"

ContextVar is better for async code because:

• It works correctly with asyncio.create_task()
• It doesn’t leak between async tasks
• It’s designed for the async execution model

Use threading.local() only when you specifically need thread-local storage.

Summary

The contextvars module provides clean context isolation for async code:

• ContextVar — Creates a context-specific variable
• .set(value) — Sets a value, returns a token for reset
• .reset(token) — Reverts to the previous value
• copy_context() — Creates a copy of the current context

This is essential for building reliable async applications where you need to maintain request-scoped state across multiple async operations.

See Also

• asyncio — The asyncio module for async programming in Python
• Async/Await Patterns in Python — Practical async/await patterns and best practices
• Threading in Python — Thread-based concurrency in Python