`ContextualFunc`: The aggregate of a callable and context managers

[thorwhalen]

An aggregate of a function and context managers.

This is useful when a function needs specific resources run, which are managed by some context managers. What ContextualFunc does is bring both in one place so that the callable is it's own context manager instance which you can enter, call, and exit.

>>> from contextlib import contextmanager
>>>
>>> def mk_test_context(name, enter_obj=None, return_instance=True):
...     @contextmanager
...     def test_context():
...         print(f'entering {name} context')
...         yield enter_obj
...         print(f'exiting {name} context')
...     return test_context()
>>> foo_context = mk_test_context('foo')
>>> bar_context = mk_test_context('bar')
>>>
>>> contextual_func = ContextualFunc(
...     lambda x: x + 1,
...     contexts=[foo_context, bar_context]
... )
>>>
>>> with contextual_func:
...     print(f"{contextual_func(2)=}")
...     print(f"{contextual_func(1414)=}")
entering foo context
entering bar context
contextual_func(2)=3
contextual_func(1414)=1415
exiting foo context
exiting bar context

┆Issue is synchronized with this Asana task by Unito

[thorwhalen]

I feel bad not re-using ContextFanout for this. I'm just not sure it's worth it for the sake of reuse.

As a matter of note, one way to use (by delegation, not subclassing) ContextFanout is to do this:

@dataclass
class ContextualFunction:
    func: Callable
    contexts: Iterable[ContextManager] = ()

    def __post_init__(self):
        self.__signature__ = Sig(self.func)
        self.multi_context = ContextFanout(*self.contexts)

    def __call__(self, *args, **kwargs):
        return self.func(*args, **kwargs)

    def __enter__(self):
        self.multi_context.__enter__()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        return self.multi_context.__exit__(exc_type, exc_val, exc_tb)