跳到主要内容

hook

TOC

Classes

🅲 Hook

class Hook(Protocol):
__HookType__: Sequence[Any] = None
__LifeSpan__: float = None
__CallInter__: int = None
__call__: Callable = None

Represents a hook that can be used to register callback functions for specific events.

This class is meant to be used with Python 3.8+ and the `typing.Protocol` type hinting feature. It defines the following class-level attributes:

  • __HookType__: A sequence of event types that this hook can handle.
  • __LifeSpan__: The duration (in seconds) that this hook will remain active once registered.
  • __CallInter__: The minimum time (in seconds) between consecutive calls to the callback function.
  • __call__: The callback function that will be executed when an event of the specified type occurs.

Note: This class cannot be instantiated directly, but it can be used to define other classes or functions that implement its interface.

🅲 MetaHookManager

class MetaHookManager(type):
stages: tuple[str, ...] = ()

🅼 __new__

__new__
def __new__(cls, name, bases, attrs):
inst = type.__new__(cls, name, bases, attrs)
stages = inst.stages
if inst.__name__ != "HookManager" and not stages:
raise HookManagerBuildError(
f"The hook manager `{inst.__name__}` must have valid stages"
)
return inst

Overrides the default `__new__` method to validate the `stages` attribute.

Returns:

  • The new instance of the `HookManager` subclass.

Raises:

🅲 HookManager

class HookManager:
stages: tuple[str, ...] = tuple()

🅼 __init__

__init__
def __init__(self, hooks: Sequence[Hook]) -> None:
assert isinstance(hooks, Sequence)
__error_msg = "The hook `{}` must have a valid `{}`, got {}"
for h in hooks:
if not hasattr(h, "__HookType__") or h.__HookType__ not in self.stages:
raise HookBuildError(
__error_msg.format(
h.__class__.__name__, "__HookType__", h.__HookType__
)
)
if not hasattr(h, "__LifeSpan__") or h.__LifeSpan__ <= 0:
raise HookBuildError(
__error_msg.format(
h.__class__.__name__, "__LifeSpan__", h.__LifeSpan__
)
)
if not hasattr(h, "__CallInter__") or h.__CallInter__ <= 0:
raise HookBuildError(
__error_msg.format(
h.__class__.__name__, "__CallInter__", h.__CallInter__
)
)
self.hooks = defaultdict(list)
self.calls: dict[str, int] = defaultdict(int)
for h in hooks:
self.hooks[h.__HookType__].append(h)

🅼 check_life_span

@staticmethod
def check_life_span(hook: Hook) -> bool:
hook.__LifeSpan__ -= 1
return hook.__LifeSpan__ <= 0

Checks whether a given `Hook` object has exceeded its maximum lifespan.

Returns:

  • True if the `Hook` object has exceeded its maximum lifespan, otherwise False.

🅼 exist

def exist(self, stage) -> bool:
return self.hooks[stage] != []

Determines whether any hooks are registered for a given event stage.

Parameters:

  • stage (str): The name of the event stage to check.

Returns:

  • True if any hooks are registered for the given event stage, otherwise False.

🅼 pre_call

def pre_call(self) -> Any:
return

Called before any hooks are executed during an event stage.

🅼 after_call

def after_call(self) -> Any:
return

Called after all hooks have been executed during an event stage.

🅼 __call__

__call__
def __call__(self, stage, *inps) -> None:
dead_hook_idx: list[int] = []
calls = self.calls[stage]
for idx, hook in enumerate(self.hooks[stage]):
if calls % hook.__CallInter__ == 0:
res = hook(*inps)
if res and self.check_life_span(hook):
dead_hook_idx.append(idx - len(dead_hook_idx))
for idx in dead_hook_idx:
self.hooks[stage].pop(idx)
self.calls[stage] = calls + 1

Executes all hooks registered for a given event stage.

Parameters:

  • stage (str): The name of the event stage to trigger.
  • *inps: Input arguments to pass to the hook functions.

🅼 call_hooks

def call_hooks(self, stage, *inps) -> None:
self.pre_call()
self(stage, *inps)
self.after_call()

Convenience method for calling all hooks at a given event stage.

Parameters:

  • stage (str): The name of the event stage to trigger.
  • *inps: Input arguments to pass to the hook functions.

🅲 ConfigHookManager

class ConfigHookManager(HookManager):
stages: tuple[str, ...] = ("pre_build", "every_build", "after_build")