Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/alex-ber/AlexBerUtils/llms.txt

Use this file to discover all available pages before exploring further.

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) -> CallableDefault: 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:
KeyTypeDescription
"var"ContextVarThe ContextVar instance.
"factory"CallableZero-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

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.

Build docs developers (and LLMs) love