hbutils.reflection.context

Thread-Local Context Variable Utilities.

This module provides a thread-safe context variable management system that allows developers to build context-dependent behavior without explicitly passing values through a call stack. It supports context inheritance, scoped variable overrides, function wrapping for new threads, nested context management, and conditional context creation.

The module contains the following main public components:

• ContextVars - Thread-local context mapping with scoped mutation helpers

• context() - Accessor for the current thread’s ContextVars

• cwrap() - Function wrapper to inherit and extend context in new threads

• nested_with() - Helper to nest multiple context managers

• conditional_with() - Conditionally enter a context manager

Note

Context instances are stored per thread and reused within the same thread. Use cwrap() to pass the current context into new threads.

Example:

>>> from contextlib import contextmanager
>>> from hbutils.reflection import context
>>>
>>> @contextmanager
... def use_mul():
...     with context().vars(mul=True):
...         yield
>>>
>>> def calc(a, b):
...     if context().get('mul', None):
...         return a * b
...     else:
...         return a + b
>>>
>>> print(calc(3, 5))
8
>>> with use_mul():
...     print(calc(3, 5))
15
>>> print(calc(3, 5))
8

__all__

hbutils.reflection.context.__all__ = ['context', 'cwrap', 'nested_with', 'conditional_with']

Built-in mutable sequence.

If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.

ContextVars

class hbutils.reflection.context.ContextVars(**kwargs: _ValueType)

Context variable management class.

This class provides a thread-safe way to manage context variables that can be temporarily modified within a with-block scope. It inherits from collections.abc.Mapping and supports standard mapping operations.

Note

This class is inherited from collections.abc.Mapping. Main features of mapping object (such as __getitem__, __len__, __iter__) are supported. See Collections Abstract Base Classes.

Warning

This object should be singleton on thread level. It is not recommended constructing manually. Use context() instead.

__getitem__(key: _KeyType) → _ValueType

Get a context variable by key.

Parameters:

key (_KeyType) – The key of the variable to retrieve.

Returns:

The value associated with the key.

Return type:

_ValueType

Raises:

KeyError – If the key is not found in the context.

__init__(**kwargs: _ValueType) → None

Initialize a ContextVars instance.

Parameters:

kwargs (_ValueType) – Initial context variable key-value pairs.

__iter__() → Iterator[_KeyType]

Iterate over the keys of context variables.

Returns:

An iterator over the context variable keys.

Return type:

Iterator[_KeyType]

__len__() → int

Get the number of variables in the context.

Returns:

The number of context variables.

Return type:

int

inherit(context_: ContextVars) → Iterator[None]

Inherit variables from another context.

This method replaces the current context variables with those from the given context. Variables not present in the given context will be removed.

Parameters:

context (ContextVars) – ContextVars object to inherit from.

Yield:

None

Note

After inherit() is used, the original variables which not present in the given ``context_`` will be removed. This is different from vars(), so attention.

vars(**kwargs: _ValueType) → Iterator[None]

Add or modify variables in the context within a with-block.

This method temporarily adds or updates context variables for the duration of the with-block. Original values are restored when exiting the block.

Parameters:

kwargs (_ValueType) – Context variables to add or modify.

Yield:

None

Examples::

>>> from hbutils.reflection import context
>>>
>>> def var_detect():
...     if context().get('var', None):
...         print(f'Var detected, its value is {context()["var"]}.')
...     else:
...         print('Var not detected.')
>>>
>>> var_detect()
Var not detected.
>>> with context().vars(var=1):
...     var_detect()
Var detected, its value is 1.
>>> var_detect()
Var not detected.

Note

See context().

context

hbutils.reflection.context.context() → ContextVars

Get the context object for the current thread.

This function returns a thread-local singleton ContextVars instance. Each thread has its own independent context that persists across function calls within that thread.

Returns:

The ContextVars object for the current thread.

Return type:

ContextVars

Note

This result is unique on one thread. Multiple calls within the same thread will return the same ContextVars instance.

cwrap

hbutils.reflection.context.cwrap(func: Callable[[...], _ValueType], *, context_: ContextVars | None = None, **vars_: _ValueType) → Callable[[...], _ValueType]

Wrap a function to inherit and extend context variables.

This decorator is essential for passing context variables into new threads, as thread-local storage is not automatically inherited by child threads.

Parameters:

• func (Callable[..., _ValueType]) – The function to wrap.

• context (Optional[ContextVars]) – Context to inherit. If None, uses the current thread’s context.

• vars (_ValueType) – Additional variables to add after inheriting the context.

Returns:

A wrapped function that executes with the inherited context.

Return type:

Callable[…, _ValueType]

Examples::

>>> from threading import Thread
>>> from hbutils.reflection import context, cwrap
>>>
>>> def var_detect():
...     if context().get('var', None):
...         print(f'Var detected, its value is {context()["var"]}.')
...     else:
...         print('Var not detected.')
>>>
>>> with context().vars(var=1):  # no inherit, vars will be lost in thread
...     t = Thread(target=var_detect)
...     t.start()
...     t.join()
Var not detected.
>>> with context().vars(var=1):  # with inherit, vars will be kept in thread
...     t = Thread(target=cwrap(var_detect))
...     t.start()
...     t.join()
Var detected, its value is 1.

Note

cwrap() is important when you need to pass the current context into thread. And it is compatible on all platforms.

Warning

cwrap() is not compatible on Windows or Python3.8+ on macOS when creating new process. Please pass in direct arguments by args argument of Process. If you insist on using context() feature, you need to pass the context object into the subprocess.

For example:

>>> from contextlib import contextmanager
>>> from multiprocessing import Process
>>> from hbutils.reflection import context
>>>
>>> @contextmanager
... def use_mul():
...     with context().vars(mul=True):
...         yield
>>>
>>> def calc(a, b):
...     if context().get('mul', None):
...         print(a * b)
...     else:
...         print(a + b)
>>>
>>> def _calc(a, b, ctx=None):
...     with context().inherit(ctx or context()):
...         return calc(a, b)
>>>
>>> if __name__ == '__main__':
...     calc(3, 5)
...     with use_mul():
...         p = Process(target=_calc, args=(3, 5, context()))
...         p.start()
...         p.join()
...     calc(3, 5)
8
15
8

nested_with

hbutils.reflection.context.nested_with(*contexts: ContextManager[Any]) → Iterator[Tuple[Any, ...]]

Enter and exit multiple context managers in a nested fashion.

This function allows you to manage multiple context managers simultaneously, entering them in order and exiting them in reverse order (LIFO).

Parameters:

contexts (ContextManager) – Variable number of context managers to nest.

Returns:

A context manager that yields a tuple of values from all nested contexts.

Return type:

ContextManager[Tuple[Any, …]]

Examples::

>>> import os.path
>>> import pathlib
>>> import tempfile
>>> from contextlib import contextmanager
>>> from hbutils.reflection import nested_with
>>>
>>> @contextmanager
... def opent(x):
...     with tempfile.TemporaryDirectory() as td:
...         pathlib.Path(os.path.join(td, f'{x}.txt')).write_text(f'this is {x}!')
...         yield td
>>>
>>> with opent(1) as d:
...     print(os.listdir(d))
...     print(pathlib.Path(f'{d}/1.txt').read_text())
['1.txt']
this is 1!
>>> with nested_with(*map(opent, range(5))) as ds:
...     for d in ds:
...         print(d)
...         print(os.path.exists(d), os.listdir(d))
...         print(pathlib.Path(f'{d}/{os.listdir(d)[0]}').read_text())
/tmp/tmp3u1984br
True ['0.txt']
this is 0!
/tmp/tmp0yx56hv0
True ['1.txt']
this is 1!
/tmp/tmpu_33drm3
True ['2.txt']
this is 2!
/tmp/tmpqal_vzgi
True ['3.txt']
this is 3!
/tmp/tmpy99_wwtt
True ['4.txt']
this is 4!
>>> for d in ds:
...     print(d)
...     print(os.path.exists(d))
/tmp/tmp3u1984br
False
/tmp/tmp0yx56hv0
False
/tmp/tmpu_33drm3
False
/tmp/tmpqal_vzgi
False
/tmp/tmpy99_wwtt
False

conditional_with

hbutils.reflection.context.conditional_with(ctx: ContextManager[Any], cond: bool) → Iterator[Any | None]

Conditionally create and enter a context manager.

This function provides a way to conditionally use a context manager based on a boolean condition. If the condition is False, the context is not entered and None is yielded instead.

Parameters:

• ctx (ContextManager) – The context manager to conditionally enter.

• cond (bool) – Boolean condition determining whether to enter the context.

Yield:

The value from the context manager if cond is True, otherwise None.

Return type:

Optional[Any]

Examples::

Here is an example of conditionally creating a temporary directory.

>>> import os.path
>>>
>>> from hbutils.reflection import conditional_with
>>> from hbutils.system import TemporaryDirectory
>>>
>>> with conditional_with(TemporaryDirectory(), cond=True) as td:
...     print('td:', td)
...     print('exist:', os.path.exists(td))
...     print('isdir:', os.path.isdir(td))
...
td: /tmp/tmp07lpb9ah
exist: True
isdir: True
>>> with conditional_with(TemporaryDirectory(), cond=False) as td:
...     print('td:', td)
...
td: None