ContextVar utilities

Collect and reset contextvars.ContextVar instances across modules with get_context_vars and reset_context_vars.

contextvars.ContextVar provides per-context (per-async-task or per-thread) state. AlexBerUtils provides two helpers for discovering all ContextVar instances in a module and resetting them to a known initial state.

get_context_vars

Scans one or more modules for top-level ContextVar instances, pairs each one with a factory function that produces its default value, resets them all immediately, and returns the list of entity dicts.

get_context_vars(*modules, factory_method_creator=None) -> List[dict]

Parameters

*modules · ModuleType · required

One or more Python module objects to scan. dir(module) is used to enumerate attributes; only those that are ContextVar instances are collected.

factory_method_creator · Callable[[ContextVar, ModuleType], Callable]

Optional function that resolves the factory callable for a given ContextVar.

Signature: (var: ContextVar, module: ModuleType) -> Callable

Default: looks for an attribute named <var.name>_DEFAULT on the same module as the ContextVar.

# Default resolution: if the ContextVar is named `request_id`,
# the factory is module.request_id_DEFAULT
factory_method_creator = lambda var, module: getattr(module, f"{var.name}_DEFAULT")

Return value

A list of dicts, each with:

Key	Type	Description
"var"	ContextVar	The ContextVar instance.
"factory"	Callable	Zero-argument callable that produces the default value.

get_context_vars calls reset_context_vars internally, so all discovered vars are reset to their defaults before the list is returned.

If no modules are passed, an empty list is returned immediately.

Usage

default factory naming

import my_module
from alexber.utils.thread_locals import get_context_vars

# my_module must expose `request_id_DEFAULT` for a ContextVar named `request_id`
entities = get_context_vars(my_module)

---

reset_context_vars

Resets every ContextVar in the entity list to its default value by calling its associated factory.

reset_context_vars(*entities) -> None

*entities · dict · required

Entity dicts as produced by get_context_vars. Each dict must have a "var" (ContextVar) and a "factory" (callable).

Raises ValueError if any factory is not callable.

from alexber.utils.thread_locals import get_context_vars, reset_context_vars
import my_module

# Collect once at startup
entities = get_context_vars(my_module)

# Reset before each request / task
reset_context_vars(*entities)

---

End-to-end example

This pattern is common in request-scoped or task-scoped applications where you want to ensure clean state at the start of every unit of work.

# my_module.py
from contextvars import ContextVar

request_id: ContextVar[str] = ContextVar('request_id')
user_id: ContextVar[int] = ContextVar('user_id')

# Factories follow the `<name>_DEFAULT` convention
def request_id_DEFAULT() -> str:
    return ''

def user_id_DEFAULT() -> int:
    return 0

# app.py
import my_module
from alexber.utils.thread_locals import get_context_vars, reset_context_vars

# At startup: discover all ContextVars and reset them
entities = get_context_vars(my_module)

async def handle_request(req):
    # Reset to defaults at the start of each request
    reset_context_vars(*entities)

    my_module.request_id.set(req.id)
    my_module.user_id.set(req.user.id)

    await process(req)

Store the entities list returned by get_context_vars as a module-level variable and reuse it across requests. Calling get_context_vars repeatedly is wasteful because it rescans the module each time.

reset_context_vars modifies the ContextVar in the current context. In an async application each asyncio.Task has its own context copy, so resetting in one task does not affect others.