Simple in-process hook registry and dispatcher. Hooks are stored under an arbitrary `hook_key`. Callers register callbacks via `register()` and trigger them via `dispatch()`.
| 5 | |
| 6 | |
| 7 | class HookManager: |
| 8 | """Simple in-process hook registry and dispatcher. |
| 9 | |
| 10 | Hooks are stored under an arbitrary `hook_key`. Callers register callbacks via `register()` |
| 11 | and trigger them via `dispatch()`. |
| 12 | """ |
| 13 | |
| 14 | def __init__(self): |
| 15 | self._hooks = defaultdict(list) |
| 16 | |
| 17 | def register(self, hook_key, func): |
| 18 | self._hooks[hook_key].append(func) |
| 19 | |
| 20 | def dispatch(self, hook_key, *args, **kwargs): |
| 21 | for func in self._hooks[hook_key]: |
| 22 | func(*args, **kwargs) |
| 23 | |
| 24 | def has(self, hook_key, func=None) -> bool: |
| 25 | """Check whether a hook key has any callbacks registered. |
| 26 | |
| 27 | Args: |
| 28 | hook_key: Hook key used during `register()` and `dispatch()`. |
| 29 | func: Optional specific callback to test membership. |
| 30 | |
| 31 | Returns: |
| 32 | True if there is at least one callback for `hook_key`, or if `func` is provided |
| 33 | and it is registered under `hook_key`. |
| 34 | """ |
| 35 | hooks = self._hooks.get(hook_key, []) |
| 36 | if func is None: |
| 37 | return bool(hooks) |
| 38 | return func in hooks |
| 39 | |
| 40 | |
| 41 | class HookStage: |