"""Core hook system implementation.""" from __future__ import annotations import logging from dataclasses import dataclass, field from enum import Enum from typing import Any, Callable, Dict, List, Optional, Protocol from .exceptions import HookExecutionError, HookRejectionError logger = logging.getLogger(__name__) class HookType(Enum): """Hook execution timing types.""" PRE = "pre" # Execute before operation (can reject) POST = "post" # Execute after operation (cannot reject, reactions only) @dataclass(frozen=True) class HookContext: """ Immutable context passed to hooks containing operation details. This is a base class that should be extended by specific implementations to add domain-specific fields. All operation parameters are frozen to prevent hooks from modifying them. Use the metadata dict to share data between hooks. Attributes: operation: The operation being performed (e.g., 'add', 'delete', 'execute') metadata: Mutable dictionary for inter-hook communication and data sharing Example: @dataclass(frozen=True) class CronjobHookContext(HookContext): job_id: str schedule: str command: str user_id: int """ operation: str # The operation being performed metadata: Dict[str, Any] = field(default_factory=dict) # Mutable bus for hooks @dataclass class PostHookError: """ Represents an error that occurred in a POST hook. POST hooks cannot reject operations, but their errors are captured and returned so they can be logged or displayed without affecting the operation's success. Attributes: hook_name: Name of the hook that raised the error error_code: Machine-readable error code (ALL_CAPS) message: Human-readable error message details: Additional error context exception: The original exception that was raised """ hook_name: str error_code: str message: str details: Dict[str, Any] exception: Exception class HookFunction(Protocol): """Protocol defining the signature for hook functions.""" def __call__(self, context: HookContext) -> bool | None: """ Execute hook logic. Args: context: Immutable operation context with mutable metadata Returns: - For PRE hooks: True to continue, False to reject, or raise HookRejectionError - For POST hooks: Return value is ignored Raises: HookRejectionError: To reject operation with error code (PRE hooks only) Exception: Any other exception is wrapped in HookExecutionError """ ... @dataclass class Hook: """ Represents a registered hook. Attributes: name: Unique identifier for the hook hook_type: When the hook executes (PRE or POST) operation: Which operation triggers this hook ('*' = all operations) callback: The function to execute priority: Execution order (lower number = higher priority) """ name: str hook_type: HookType operation: str # The operation this hook applies to, or '*' for all callback: HookFunction priority: int = 100 # Lower number = higher priority class HookRegistry: """ Registry for managing hooks with priority support. Hooks are stored per operation type and executed by priority (lower = higher priority). Example: registry = HookRegistry(operations=['schedule', 'execute', 'pause', 'delete']) registry.register( name='check_permissions', hook_type=HookType.PRE, callback=check_permissions_func, operation='delete', priority=10 ) """ def __init__(self, operations: Optional[List[str]] = None): """ Initialize hook registry. Args: operations: List of valid operation names. If None, defaults to ['*'] """ self._hooks: Dict[str, List[Hook]] = {"*": []} # Global hooks for all operations if operations: for op in operations: self._hooks[op] = [] def register( self, name: str, hook_type: HookType, callback: HookFunction, operation: str = "*", priority: int = 100, ) -> None: """ Register a new hook. Args: name: Unique identifier for the hook hook_type: HookType.PRE or HookType.POST callback: Function to execute operation: Operation name or '*' for all operations priority: Execution priority (lower number = higher priority, default=100) Raises: ValueError: If operation is invalid or hook name already exists Example: registry.register( name="check_daily_limit", hook_type=HookType.PRE, callback=check_daily_limit_hook, operation="execute", priority=50, ) """ if operation not in self._hooks: raise ValueError( f"Invalid operation: {operation}. Must be one of: {', '.join(self._hooks.keys())}" ) # Check for duplicate names across all operations for hooks in self._hooks.values(): if any(h.name == name for h in hooks): raise ValueError(f"Hook with name '{name}' already registered") hook = Hook( name=name, hook_type=hook_type, operation=operation, callback=callback, priority=priority, ) self._hooks[operation].append(hook) # Sort by priority (lower number = higher priority), then by registration order self._hooks[operation].sort(key=lambda h: (h.priority, h.name)) logger.info( f"Registered {hook_type.value} hook '{name}' for operation '{operation}' with priority {priority}" ) def unregister(self, name: str) -> bool: """ Unregister a hook by name. Args: name: Hook name to remove Returns: True if hook was found and removed, False otherwise """ for operation, hooks in self._hooks.items(): for hook in hooks: if hook.name == name: hooks.remove(hook) logger.info(f"Unregistered hook '{name}' from operation '{operation}'") return True return False def get_hooks(self, operation: str, hook_type: HookType) -> List[Hook]: """ Get all hooks for a specific operation and type, sorted by priority. Args: operation: Operation name hook_type: HookType.PRE or HookType.POST Returns: List of hooks sorted by priority (lower number first) """ # Get operation-specific hooks + global hooks operation_hooks = self._hooks.get(operation, []) global_hooks = self._hooks.get("*", []) # Combine and filter by hook type all_hooks = operation_hooks + global_hooks filtered = [h for h in all_hooks if h.hook_type == hook_type] # Sort by priority, then name filtered.sort(key=lambda h: (h.priority, h.name)) return filtered def clear(self) -> None: """Clear all registered hooks (useful for testing).""" for operation in self._hooks: self._hooks[operation].clear() logger.info("Cleared all hooks") class HookManager: """ Manages hook execution for operations. Executes PRE hooks before operation (can reject) and POST hooks after (cannot reject). Example: manager = HookManager(registry) # Before operation context = MyHookContext(operation='delete', user_id=123) try: manager.execute_pre_hooks(context) except HookRejectionError as e: return {"error": e.error_code, "message": e.message} # Perform operation result = perform_operation() # After operation errors = manager.execute_post_hooks(context) return {"success": True, "post_hook_errors": errors} """ def __init__(self, registry: Optional[HookRegistry] = None): """ Initialize hook manager. Args: registry: Hook registry to use. If None, uses global registry. """ self.registry = registry or HookRegistry() def execute_pre_hooks(self, context: HookContext) -> None: """ Execute all PRE hooks for the given operation. PRE hooks can reject the operation by: 1. Returning False 2. Raising HookRejectionError with error_code and message Args: context: Operation context Raises: HookRejectionError: If any hook rejects the operation HookExecutionError: If a hook fails unexpectedly """ hooks = self.registry.get_hooks(context.operation, HookType.PRE) for hook in hooks: logger.debug(f"Executing PRE hook '{hook.name}' for {context.operation}") try: result = hook.callback(context) # Check if hook rejected operation if result is False: raise HookRejectionError( error_code=f"HOOK_REJECTED_{hook.name.upper()}", message=f"Operation rejected by hook: {hook.name}", details={"hook_name": hook.name}, ) logger.debug(f"PRE hook '{hook.name}' completed successfully") except HookRejectionError: # Re-raise rejection errors as-is raise except Exception as e: # Wrap unexpected errors logger.error(f"PRE hook '{hook.name}' failed with error: {e}", exc_info=True) raise HookExecutionError( message=f"Hook '{hook.name}' failed to execute: {str(e)}", hook_name=hook.name, original_exception=e, ) from e def execute_post_hooks(self, context: HookContext) -> List[PostHookError]: """ Execute all POST hooks for the given operation. POST hooks cannot reject the operation. Errors are captured and returned so the operation remains successful but errors can be logged/displayed. Args: context: Operation context (typically with updated state after operation) Returns: List of errors that occurred in POST hooks (empty if all succeeded) """ hooks = self.registry.get_hooks(context.operation, HookType.POST) errors: List[PostHookError] = [] for hook in hooks: logger.debug(f"Executing POST hook '{hook.name}' for {context.operation}") try: hook.callback(context) logger.debug(f"POST hook '{hook.name}' completed successfully") except HookRejectionError as e: # POST hooks trying to reject - capture as error but don't fail operation logger.warning( f"POST hook '{hook.name}' tried to reject operation: {e.error_code} - {e.message}" ) errors.append( PostHookError( hook_name=hook.name, error_code=e.error_code, message=e.message, details=e.details, exception=e, ) ) except Exception as e: # Capture all other errors logger.error(f"POST hook '{hook.name}' failed with error: {e}", exc_info=True) errors.append( PostHookError( hook_name=hook.name, error_code="HOOK_EXECUTION_ERROR", message=f"Hook '{hook.name}' failed: {str(e)}", details={"exception_type": type(e).__name__}, exception=e, ) ) return errors # Global hook registry singleton _global_registry: Optional[HookRegistry] = None def get_global_registry() -> HookRegistry: """ Get the global hook registry. Lazily initializes the global registry on first access. """ global _global_registry if _global_registry is None: _global_registry = HookRegistry() return _global_registry def register_hook( name: str, hook_type: HookType, callback: HookFunction, operation: str = "*", priority: int = 100, ) -> None: """ Register a hook in the global registry. Args: name: Unique identifier for the hook hook_type: HookType.PRE or HookType.POST callback: Function to execute operation: Operation name or '*' for all operations priority: Execution priority (lower number = higher priority, default=100) Example: def check_quota(context: HookContext) -> bool: if context.metadata.get('count', 0) > 100: raise HookRejectionError( error_code="QUOTA_EXCEEDED", message="You have exceeded your quota limit" ) return True register_hook( name="quota_check", hook_type=HookType.PRE, callback=check_quota, operation="create", priority=10, ) """ get_global_registry().register(name, hook_type, callback, operation, priority) def unregister_hook(name: str) -> bool: """ Unregister a hook from the global registry. Args: name: Hook name to remove Returns: True if hook was found and removed, False otherwise """ return get_global_registry().unregister(name) def clear_hooks() -> None: """Clear all hooks from the global registry (useful for testing).""" get_global_registry().clear()