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

nested_with

hbutils.reflection.context.nested_with(*contexts) → AbstractContextManager[Tuple[Any, …]]

Overview:

Nested with, enter and exit multiple contexts.

Parameters:

contexts – Contexts to manage.

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)

Overview:

Conditional create context.

Parameters:

• ctx – Context object.

• cond – Condition for create or not.

Examples::

Here is an example of conditionally create 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