hbutils.reflection.context

Overview:

Utilities for building context variables on thread level.

This is useful when implementing a with-block-based syntax. For 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)

Overview:

Context variable management.

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.

__init__(**kwargs)

Constructor of ContextVars.

Parameters:

kwargs – Initial context values.

inherit(context_: hbutils.reflection.context.ContextVars)

Inherit variables from the given context.

Parameters:

context – ContextVars object to inherit from.

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)

Adding variables into context of with block.

Parameters:

kwargs – Additional context variables.

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() → hbutils.reflection.context.ContextVars

Overview:

Get context object in this thread.

Returns:

Context object in this thread.

Note

This result is unique on one thread.

cwrap

hbutils.reflection.context.cwrap(func, *, context_: Optional[hbutils.reflection.context.ContextVars] = None, **vars_)

Overview:

Context wrap for functions.

Parameters:

• func – Original function to wrap.

• context – Context for inheriting. Default is None which means context()’s result will be used.

• vars – External variables after inherit context.

Note

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

For example:

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

Warning

cwrap() is not compitable 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