Context Variables (contextvars)

contextvars provides context-local state that works correctly with both threads and async tasks.

Why contextvars?

The Problem with threading.local in Async

import threading
import asyncio

# threading.local() doesn't work properly with async
local = threading.local()

async def task(name):
    local.name = name
    await asyncio.sleep(0.1)
    # BUG: May see wrong value after await!
    print(f"Task: {local.name}")

async def main():
    await asyncio.gather(
        task("task1"),
        task("task2")
    )

# Both might print "task2" because they share the same thread

The Solution: contextvars

import contextvars
import asyncio

# contextvars works correctly with async
name_var = contextvars.ContextVar('name')

async def task(name):
    name_var.set(name)
    await asyncio.sleep(0.1)
    # Correct: Each task sees its own value
    print(f"Task: {name_var.get()}")

async def main():
    await asyncio.gather(
        task("task1"),
        task("task2")
    )
    # Output: "task1" and "task2" correctly

asyncio.run(main())

Basic Usage

Creating and Using ContextVar

import contextvars

# Create a context variable
request_id = contextvars.ContextVar('request_id', default=None)

# Set value
request_id.set('req-123')

# Get value
print(request_id.get())  # 'req-123'

# Get with default (if not set)
other_var = contextvars.ContextVar('other')
print(other_var.get('default'))  # 'default'
print(other_var.get())  # Raises LookupError if no default

Token-Based Reset

import contextvars

var = contextvars.ContextVar('var', default='initial')

# set() returns a token
token = var.set('new_value')
print(var.get())  # 'new_value'

# Reset to previous value using token
var.reset(token)
print(var.get())  # 'initial'

Context Management

Working with Context Objects

import contextvars

var = contextvars.ContextVar('var')

# Get current context
ctx = contextvars.copy_context()

# Run function in context
def show_value():
    return var.get('not set')

# Set in current context
var.set('hello')
print(ctx.run(show_value))  # 'not set' (ctx was copied before set)

# Create new context with modifications
ctx2 = contextvars.copy_context()
print(ctx2.run(show_value))  # 'hello'

Running in Isolated Context

import contextvars

request_id = contextvars.ContextVar('request_id')

def process_request(req_id):
    request_id.set(req_id)
    # ... do work ...
    return request_id.get()

# Run in isolated context
ctx = contextvars.copy_context()
result = ctx.run(process_request, 'req-123')

# Original context unaffected
print(request_id.get('not set'))  # 'not set'

Async Task Context

Automatic Context Copying

import asyncio
import contextvars

user_id = contextvars.ContextVar('user_id')

async def get_user_data():
    # Each task has its own copy of context
    uid = user_id.get()
    await asyncio.sleep(0.1)
    return f"Data for user {uid}"

async def handle_request(uid):
    user_id.set(uid)
    # Task created here inherits context
    task = asyncio.create_task(get_user_data())
    return await task

async def main():
    results = await asyncio.gather(
        handle_request('user-1'),
        handle_request('user-2')
    )
    print(results)
    # ['Data for user user-1', 'Data for user user-2']

asyncio.run(main())

Context in Callbacks

import asyncio
import contextvars

var = contextvars.ContextVar('var')

async def main():
    var.set('main_value')

    loop = asyncio.get_running_loop()

    def callback():
        # Callback runs in context where it was scheduled
        print(f"Callback: {var.get()}")

    # Schedule callback - captures current context
    loop.call_soon(callback)

    await asyncio.sleep(0.1)

asyncio.run(main())  # Prints: "Callback: main_value"

Practical Examples

1. Request Context in Web Framework

import contextvars
from contextlib import contextmanager

# Context variables for request
request_id = contextvars.ContextVar('request_id')
current_user = contextvars.ContextVar('current_user')

@contextmanager
def request_context(req_id, user):
    """Set request context for the duration of handling."""
    token_id = request_id.set(req_id)
    token_user = current_user.set(user)
    try:
        yield
    finally:
        request_id.reset(token_id)
        current_user.reset(token_user)

def get_request_id():
    return request_id.get(None)

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

# Usage in async handler
async def handle_request(req_id, user):
    with request_context(req_id, user):
        # All code here sees the context
        result = await process_request()
        log_request()  # Can access request_id
        return result

def log_request():
    print(f"Request {get_request_id()} by {get_current_user()}")

2. Database Transaction Context

import contextvars
from contextlib import asynccontextmanager

_transaction = contextvars.ContextVar('transaction', default=None)

@asynccontextmanager
async def transaction(conn):
    """Provide transaction context."""
    tx = await conn.begin()
    token = _transaction.set(tx)
    try:
        yield tx
        await tx.commit()
    except Exception:
        await tx.rollback()
        raise
    finally:
        _transaction.reset(token)

def get_transaction():
    """Get current transaction or raise."""
    tx = _transaction.get()
    if tx is None:
        raise RuntimeError("No active transaction")
    return tx

async def save_user(user):
    tx = get_transaction()
    await tx.execute("INSERT INTO users ...")

3. Logging Context

import contextvars
import logging

# Context for logging
log_context = contextvars.ContextVar('log_context', default={})

class ContextFilter(logging.Filter):
    def filter(self, record):
        ctx = log_context.get()
        for key, value in ctx.items():
            setattr(record, key, value)
        return True

def with_log_context(**kwargs):
    """Add context to all logs in this scope."""
    current = log_context.get().copy()
    current.update(kwargs)
    return log_context.set(current)

# Usage
async def handle_request(request_id, user_id):
    token = with_log_context(request_id=request_id, user_id=user_id)
    try:
        logger.info("Processing request")  # Includes context
        await do_work()
        logger.info("Request completed")
    finally:
        log_context.reset(token)

4. Timeout Context

import contextvars
import asyncio
import time

deadline = contextvars.ContextVar('deadline', default=None)

def get_remaining_time():
    """Get remaining time until deadline."""
    dl = deadline.get()
    if dl is None:
        return float('inf')
    return max(0, dl - time.time())

@asynccontextmanager
async def timeout(seconds):
    """Set deadline for operations."""
    new_deadline = time.time() + seconds
    current = deadline.get()

    # Use earliest deadline
    if current is not None:
        new_deadline = min(new_deadline, current)

    token = deadline.set(new_deadline)
    try:
        yield
    finally:
        deadline.reset(token)

async def fetch_with_deadline(url):
    remaining = get_remaining_time()
    if remaining <= 0:
        raise asyncio.TimeoutError("Deadline exceeded")

    async with aiohttp.ClientSession() as session:
        async with asyncio.timeout(remaining):
            async with session.get(url) as r:
                return await r.text()

Comparison: threading.local vs contextvars

Feature	threading.local	contextvars
Thread isolation	✅	✅
Async task isolation	❌	✅
Automatic context copy	❌	✅
Reset to previous value	❌	✅ (tokens)
Explicit context objects	❌	✅
Python version	2.4+	3.7+

Best Practices

1. Use Meaningful Names

# Good: Descriptive names
request_id = contextvars.ContextVar('request_id')
current_user = contextvars.ContextVar('current_user')

# Bad: Generic names
var1 = contextvars.ContextVar('var1')

2. Provide Defaults When Appropriate

# With default
log_level = contextvars.ContextVar('log_level', default='INFO')

# Without default (requires explicit set)
auth_token = contextvars.ContextVar('auth_token')

3. Always Reset in Finally

token = var.set('value')
try:
    # ... do work ...
finally:
    var.reset(token)

# Or use context manager
@contextmanager
def scoped_var(var, value):
    token = var.set(value)
    try:
        yield
    finally:
        var.reset(token)

4. Document Context Dependencies

def process_order():
    """Process the current order.

    Requires:
        - current_user context variable to be set
        - request_id context variable to be set
    """
    user = current_user.get()  # Raises if not set
    req_id = request_id.get()

Key Takeaways

• Use contextvars for state that should be isolated per async task
• Each asyncio.create_task() gets a copy of the current context
• Use tokens from set() to reset() values properly
• Works for both sync and async code
• Replaces threading.local() for async-aware context
• Essential for request context, transactions, logging in async apps