hbutils.reflection.context

Overview:

Utilities for building context variables on thread level.

This module provides a thread-safe context variable management system that allows developers to create with-block-based syntax for managing state across function calls. It’s particularly useful for implementing context-dependent behavior without explicitly passing parameters through the call stack.

The main features include:

• Thread-level context variable storage

• Context inheritance and variable scoping

• Context wrapping for functions (useful in threading)

• Nested context management

• Conditional context creation

Example::

>>> from contextlib import contextmanager
>>> from hbutils.reflection import context
>>>
>>> # developer's view
... @contextmanager
... def use_mul():  # set 'mul' to `True` in its with-block
...     with context().vars(mul=True):
...         yield
>>>
>>> def calc(a, b):  # logic of `calc` will be changed when 'mul' is given
...     if context().get('mul', None):
...         return a * b
...     else:
...         return a + b
>>>
>>> # user's view (magic-liked, isn't it?)
... print(calc(3, 5))  # 3 + 5
8
>>> with use_mul():
...     print(calc(3, 5))  # changed to 3 * 5
15
>>> print(calc(3, 5))  # back to 3 + 5, again :)
8

ContextVars

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

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)

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)

Initialize a ContextVars instance.

Parameters:

kwargs (Any) – 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)

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)

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 (Any) – 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, *, context_: ContextVars | None = None, **vars_)

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) – The function to wrap.

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

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

Returns:

A wrapped function that executes with the inherited context.

Return type:

callable

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 sub process.

For example:

>>> from contextlib import contextmanager
>>> from multiprocessing import Process
>>> from hbutils.reflection import context
>>>
>>> # developer's view
... @contextmanager
... def use_mul():  # set 'mul' to `True` in its with-block
...     with context().vars(mul=True):
...         yield
>>>
>>> def calc(a, b):  # logic of `calc` will be changed when 'mul' is given
...     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)
>>>
>>> # user's view
... if __name__ == '__main__':
...     calc(3, 5)  # 3 + 5
...     with use_mul():
...         p = Process(target=_calc, args=(3, 5, context()))
...         p.start()
...         p.join()
...     calc(3, 5)  # back to 3 + 5, again :)
8
15
8

nested_with

hbutils.reflection.context.nested_with(*contexts) → ContextManager[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
>>>
>>> # allocate a temporary directory, and put one file inside
>>> @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
>>>
>>> # let's try it
>>> with opent(1) as d:
...     print(os.listdir(d))
...     print(pathlib.Path(f'{d}/1.txt').read_text())
['1.txt']
this is 1!
>>> # open 5 temporary directories at one time
>>> 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!
>>> # these directories are released now
>>> for d in ds:
...     print(d)
...     print(os.path.exists(d))  # not exist anymore
/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, cond)

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:

Any or None

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
>>>
>>> # create
>>> 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
>>> # not create
>>> with conditional_with(TemporaryDirectory(), cond=False) as td:
...     print('td:', td)
...
td: None