core_daemon/hook_manager.py

111 lines
4.0 KiB
Python
Raw Normal View History

"""
hook_manager.py
This module provides the HookManager class which allows for the registration,
unregistration, and execution of hooks. Hooks are functions that can be executed
at specific points in the program flow, such as before or after commands or module
loading/unloading.
Classes:
HookManager: Manages the registration, unregistration, and execution of hooks.
Variables:
hook_manager (HookManager): An instance of HookManager for global usage.
"""
import logging
from typing import Callable, List, Dict
class HookManager:
"""
A manager for handling hooks that can be registered, unregistered, and executed
at various points in a program's flow.
Attributes:
hooks (dict): A dictionary containing lists of functions for each hook point.
"""
def __init__(self):
"""
Initializes a new instance of the HookManager class with predefined hook points.
"""
self.hooks: Dict[str, List[Callable]] = {
'pre_command': [],
'post_command': [],
'pre_module_load': [],
'post_module_load': [],
'pre_module_unload': [],
'post_module_unload': [],
}
logging.info("HookManager initialized with hooks: %s", list(self.hooks.keys()))
def register_hook(self, hook_name: str, func: Callable):
"""
Registers a function to a specified hook.
Args:
hook_name (str): The name of the hook to register the function to.
func (function): The function to register.
Raises:
ValueError: If the specified hook name is not recognized.
"""
if hook_name in self.hooks:
self.hooks[hook_name].append(func)
logging.info("Registered function %s to hook %s", func.__name__, hook_name)
else:
logging.error("Attempted to register to unknown hook: %s", hook_name)
raise ValueError(f"Unknown hook: {hook_name}")
def unregister_hook(self, hook_name: str, func: Callable):
"""
Unregisters a function from a specified hook.
Args:
hook_name (str): The name of the hook to unregister the function from.
func (function): The function to unregister.
Raises:
ValueError: If the specified hook name is not recognized.
"""
if hook_name in self.hooks:
if func in self.hooks[hook_name]:
self.hooks[hook_name].remove(func)
logging.info("Unregistered function %s from hook %s", func.__name__, hook_name)
else:
logging.warning("Function %s not found in hook %s", func.__name__, hook_name)
else:
logging.error("Attempted to unregister from unknown hook: %s", hook_name)
raise ValueError(f"Unknown hook: {hook_name}")
def execute_hook(self, hook_name: str, *args, **kwargs) -> List:
"""
Executes all functions registered to a specified hook with the provided arguments.
Args:
hook_name (str): The name of the hook to execute the functions for.
*args: Variable length argument list to pass to the hook functions.
**kwargs: Arbitrary keyword arguments to pass to the hook functions.
Returns:
list: A list of results from each hook function executed.
Raises:
ValueError: If the specified hook name is not recognized.
"""
if hook_name not in self.hooks:
logging.error("Attempted to execute unknown hook: %s", hook_name)
raise ValueError(f"Unknown hook: {hook_name}")
results = []
for func in self.hooks[hook_name]:
try:
result = func(*args, **kwargs)
results.append(result)
logging.info("Executed function %s in hook %s", func.__name__, hook_name)
except Exception as e:
logging.error("Error executing function %s in hook %s: %s", func.__name__, hook_name, str(e))
return results
hook_manager = HookManager()