# -*- coding: utf-8 -*- """ Hook-Registry für zentrale Hook-Verwaltung. Ermöglicht das Registrieren, Finden und Verwalten von Hooks über alle Plugins hinweg. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, TypeVar from trixy_core.plugins.hooks.hook import ( PluginHook, HookPriority, HookCallback, HookHandler, ) T = TypeVar("T") @dataclass class HookInfo: """ Informationen über einen registrierten Hook. Attributes: name: Hook-Name description: Beschreibung handler_count: Anzahl Handler total_calls: Gesamtaufrufe handlers: Handler-Details """ name: str description: str handler_count: int total_calls: int handlers: list[dict[str, Any]] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: """Konvertiert zu Dictionary.""" return { "name": self.name, "description": self.description, "handler_count": self.handler_count, "total_calls": self.total_calls, "handlers": self.handlers, } class HookRegistry: """ Zentrale Registry für alle Plugin-Hooks. Verwaltet Hooks und ermöglicht deren dynamische Erstellung und Abfrage. Example: registry = HookRegistry() # Hook erstellen oder abrufen process_hook = registry.get_or_create("process_text") # Handler registrieren registry.register_handler( "process_text", "my_plugin", lambda text: text.upper() ) # Hook anwenden result = registry.apply("process_text", "hello") print(result.value) # "HELLO" # Alle Hooks eines Plugins entfernen registry.remove_plugin("my_plugin") # Statistiken for info in registry.get_all_info(): print(f"{info.name}: {info.handler_count} handlers") """ def __init__(self) -> None: """Initialisiert die Registry.""" self._hooks: dict[str, PluginHook] = {} self._plugin_hooks: dict[str, set[str]] = {} # plugin_name -> set(hook_names) @property def hook_count(self) -> int: """Anzahl registrierter Hooks.""" return len(self._hooks) @property def total_handlers(self) -> int: """Gesamtanzahl aller Handler.""" return sum(hook.handler_count for hook in self._hooks.values()) def create( self, name: str, description: str = "", default_value: Any = None, ) -> PluginHook: """ Erstellt einen neuen Hook. Args: name: Eindeutiger Hook-Name description: Beschreibung default_value: Standardwert Returns: Der erstellte Hook Raises: ValueError: Wenn Hook bereits existiert """ if name in self._hooks: raise ValueError(f"Hook existiert bereits: {name}") hook: PluginHook = PluginHook(name, description, default_value) self._hooks[name] = hook return hook def get(self, name: str) -> PluginHook | None: """Gibt einen Hook zurück.""" return self._hooks.get(name) def get_or_create( self, name: str, description: str = "", default_value: Any = None, ) -> PluginHook: """ Gibt einen Hook zurück oder erstellt ihn. Args: name: Hook-Name description: Beschreibung (nur bei Erstellung) default_value: Standardwert (nur bei Erstellung) Returns: Der Hook """ if name not in self._hooks: self._hooks[name] = PluginHook(name, description, default_value) return self._hooks[name] def remove(self, name: str) -> PluginHook | None: """ Entfernt einen Hook. Args: name: Hook-Name Returns: Der entfernte Hook oder None """ hook = self._hooks.pop(name, None) # Aus Plugin-Mapping entfernen for hooks in self._plugin_hooks.values(): hooks.discard(name) return hook def register_handler( self, hook_name: str, plugin_name: str, callback: HookCallback, priority: HookPriority = HookPriority.NORMAL, once: bool = False, ) -> bool: """ Registriert einen Handler für einen Hook. Erstellt den Hook automatisch wenn nicht vorhanden. Args: hook_name: Hook-Name plugin_name: Plugin-Name callback: Callback-Funktion priority: Ausführungspriorität once: Nur einmal ausführen Returns: True bei Erfolg """ hook = self.get_or_create(hook_name) hook.add_handler(callback, plugin_name, priority, once) # Plugin-Mapping aktualisieren if plugin_name not in self._plugin_hooks: self._plugin_hooks[plugin_name] = set() self._plugin_hooks[plugin_name].add(hook_name) return True def unregister_handler( self, hook_name: str, plugin_name: str, ) -> int: """ Entfernt Handler eines Plugins von einem Hook. Args: hook_name: Hook-Name plugin_name: Plugin-Name Returns: Anzahl entfernter Handler """ hook = self._hooks.get(hook_name) if hook is None: return 0 removed = hook.remove_handler(plugin_name) # Plugin-Mapping aktualisieren if plugin_name in self._plugin_hooks: self._plugin_hooks[plugin_name].discard(hook_name) return removed def remove_plugin(self, plugin_name: str) -> int: """ Entfernt alle Handler eines Plugins. Args: plugin_name: Plugin-Name Returns: Gesamtanzahl entfernter Handler """ total_removed = 0 if plugin_name in self._plugin_hooks: for hook_name in list(self._plugin_hooks[plugin_name]): hook = self._hooks.get(hook_name) if hook: total_removed += hook.remove_handler(plugin_name) del self._plugin_hooks[plugin_name] return total_removed def apply( self, hook_name: str, value: Any, *args: Any, **kwargs: Any, ) -> Any: """ Wendet einen Hook an. Args: hook_name: Hook-Name value: Zu verarbeitender Wert *args: Zusätzliche Argumente **kwargs: Keyword-Argumente Returns: HookResult oder der Originalwert """ hook = self._hooks.get(hook_name) if hook is None: from trixy_core.plugins.hooks.hook import HookResult return HookResult(value=value) return hook.apply(value, *args, **kwargs) async def apply_async( self, hook_name: str, value: Any, *args: Any, **kwargs: Any, ) -> Any: """Wendet einen Hook asynchron an.""" hook = self._hooks.get(hook_name) if hook is None: from trixy_core.plugins.hooks.hook import HookResult return HookResult(value=value) return await hook.apply_async(value, *args, **kwargs) def trigger( self, hook_name: str, *args: Any, **kwargs: Any, ) -> list[Any]: """ Triggert einen Hook als Action. Args: hook_name: Hook-Name *args: Argumente **kwargs: Keyword-Argumente Returns: Liste der Ergebnisse """ hook = self._hooks.get(hook_name) if hook is None: return [] return hook.trigger(*args, **kwargs) async def trigger_async( self, hook_name: str, *args: Any, **kwargs: Any, ) -> list[Any]: """Triggert einen Hook asynchron.""" hook = self._hooks.get(hook_name) if hook is None: return [] return await hook.trigger_async(*args, **kwargs) def get_hooks_for_plugin(self, plugin_name: str) -> list[str]: """ Gibt alle Hooks zurück, für die ein Plugin Handler hat. Args: plugin_name: Plugin-Name Returns: Liste der Hook-Namen """ return list(self._plugin_hooks.get(plugin_name, set())) def get_plugins_for_hook(self, hook_name: str) -> list[str]: """ Gibt alle Plugins zurück, die Handler für einen Hook haben. Args: hook_name: Hook-Name Returns: Liste der Plugin-Namen """ hook = self._hooks.get(hook_name) if hook is None: return [] return list(set(h.plugin_name for h in hook.get_handlers())) def get_info(self, hook_name: str) -> HookInfo | None: """ Gibt Informationen über einen Hook zurück. Args: hook_name: Hook-Name Returns: HookInfo oder None """ hook = self._hooks.get(hook_name) if hook is None: return None handlers = [] total_calls = 0 for handler in hook.get_handlers(): total_calls += handler.call_count handlers.append({ "plugin": handler.plugin_name, "priority": handler.priority.name, "is_async": handler.is_async, "enabled": handler.enabled, "call_count": handler.call_count, }) return HookInfo( name=hook.name, description=hook.description, handler_count=hook.handler_count, total_calls=total_calls, handlers=handlers, ) def get_all_info(self) -> list[HookInfo]: """Gibt Informationen über alle Hooks zurück.""" return [ info for name in self._hooks if (info := self.get_info(name)) is not None ] def list_hooks(self) -> list[str]: """Gibt alle Hook-Namen zurück.""" return list(self._hooks.keys()) def has_hook(self, name: str) -> bool: """Prüft, ob ein Hook existiert.""" return name in self._hooks def clear(self) -> None: """Entfernt alle Hooks und Handler.""" self._hooks.clear() self._plugin_hooks.clear() def __contains__(self, name: str) -> bool: return name in self._hooks def __getitem__(self, name: str) -> PluginHook: if name not in self._hooks: raise KeyError(f"Hook nicht gefunden: {name}") return self._hooks[name] def __len__(self) -> int: return len(self._hooks)