`@hook` decorator#

`hook`([inp, enabled, priority])	Initialize hook from callable/context manager.
`HookPriority`(value[, names, module, ...])	Hook priority enum.
`Hook`(callback[, enabled, priority])	Hook representation.
`Hook.enable`()	Enable the hook.
`Hook.disable`()	Disable the hook.
`Hook.skip`()	Temporary disable the hook.

@onetl.hooks.hook.hook(inp: Callable[[...], T] | None = None, enabled: bool = True, priority: HookPriority = HookPriority.NORMAL)#

Initialize hook from callable/context manager.

Examples

from onetl.hooks import hook, HookPriority

@hook
def some_func(*args, **kwargs):
    ...

@hook(enabled=True, priority=HookPriority.FIRST)
def another_func(*args, **kwargs):
    ...

from onetl.hooks import hook, HookPriority

@hook
class SimpleContextManager:
    def __init__(self, *args, **kwargs):
        ...

    def __enter__(self):
        ...
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        ...
        return False

@hook(enabled=True, priority=HookPriority.FIRST)
class ContextManagerWithProcessResult:
    def __init__(self, *args, **kwargs):
        ...

    def __enter__(self):
        ...
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        ...
        return False

    def process_result(self, result):
        # special method to handle method result call
        return modify(result)

    ...

class onetl.hooks.hook.HookPriority(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)#

Hook priority enum.

All hooks within the same priority are executed in the same order they were registered.

FIRST = -1#: Hooks with this priority will run first.

NORMAL = 0#: Hooks with this priority will run after FIRST but before LAST.

LAST = 1#: Hooks with this priority will run last.

class onetl.hooks.hook.Hook(callback: Callable[[...], T], enabled: bool = True, priority: HookPriority = HookPriority.NORMAL)#

Hook representation.

Parameters:

callbacktyping.Callable: Some callable object which will be wrapped into a Hook, like function or ContextManager class.
enabledbool: Will hook be executed or not. Useful for debugging.
priorityonetl.hooks.hook.HookPriority: Changes hooks priority, see HookPriority documentation.

Examples

from onetl.hooks.hook import Hook, HookPriority

def some_func(*args, **kwargs): ...

hook = Hook(callback=some_func, enabled=True, priority=HookPriority.FIRST)

enable()#

Enable the hook.

Examples

hook = Hook(..., enabled=False)
assert not hook.enabled

hook.enable()
assert hook.enabled

disable()#

Disable the hook.

Examples

hook = Hook(..., enabled=True)
assert hook.enabled

hook.disable()
assert not hook.enabled

skip()#

Temporary disable the hook.

Note

If hook was created with enabled=False, or was disabled by disable, its state will left intact after exiting the context.

You should call enable explicitly to change its state.

Examples

hook = Hook(..., enabled=True)
assert hook.enabled

with hook.skip():
    assert not hook.enabled

# hook state is restored as it was before entering the context manager
assert hook.enabled

hook = Hook(..., enabled=True)
assert hook.enabled

@hook.skip()
def hook_disabled():
    assert not hook.enabled

hook_disabled()

# hook state is restored as it was before entering the context manager
assert hook.enabled

@hook decorator#

`@hook` decorator#