# -*- coding: utf-8 -*- """ Plugin-Hook Implementierung. Hooks ermöglichen Plugins, sich in definierte Erweiterungspunkte einzuklinken und Daten zu modifizieren. """ from __future__ import annotations import asyncio from dataclasses import dataclass, field from datetime import datetime from enum import IntEnum, auto from typing import Any, Callable, TypeVar, Generic T = TypeVar("T") # Hook-Callback-Typen HookCallback = Callable[..., Any] AsyncHookCallback = Callable[..., Any] # Koroutine class HookPriority(IntEnum): """Priorität für Hook-Callbacks.""" FIRST = 0 # Wird zuerst ausgeführt HIGH = 25 NORMAL = 50 # Standard LOW = 75 LAST = 100 # Wird zuletzt ausgeführt @dataclass class HookResult(Generic[T]): """ Ergebnis eines Hook-Aufrufs. Attributes: value: Der Ergebniswert modified: Wurde der Wert modifiziert stopped: Wurde die Verarbeitung gestoppt source: Welches Plugin hat modifiziert errors: Aufgetretene Fehler """ value: T modified: bool = False stopped: bool = False source: str = "" errors: list[str] = field(default_factory=list) @property def success(self) -> bool: """Keine Fehler aufgetreten.""" return len(self.errors) == 0 @dataclass class HookHandler: """ Ein registrierter Hook-Handler. Attributes: callback: Die Callback-Funktion plugin_name: Name des registrierenden Plugins priority: Ausführungspriorität is_async: Ist asynchron once: Nur einmal ausführen enabled: Ist aktiviert call_count: Anzahl Aufrufe """ callback: HookCallback plugin_name: str priority: HookPriority = HookPriority.NORMAL is_async: bool = False once: bool = False enabled: bool = True call_count: int = 0 created_at: datetime = field(default_factory=datetime.now) def __lt__(self, other: HookHandler) -> bool: """Sortierung nach Priorität.""" return self.priority < other.priority class PluginHook(Generic[T]): """ Ein Plugin-Hook für Erweiterungspunkte. Hooks ermöglichen Plugins, sich in definierte Punkte einzuklinken und Daten zu verarbeiten oder zu modifizieren. Es gibt zwei Hauptmuster: 1. Filter: Modifiziert einen Wert (z.B. Text-Transformation) 2. Action: Führt eine Aktion aus (z.B. Logging) Example: # Hook definieren process_text = PluginHook[str]("process_text") # Handler registrieren @process_text.register("my_plugin") def uppercase(text: str) -> str: return text.upper() @process_text.register("other_plugin", priority=HookPriority.LAST) def add_period(text: str) -> str: if not text.endswith("."): return text + "." return text # Hook ausführen result = process_text.apply("hello world") print(result.value) # "HELLO WORLD." # Async-Unterstützung @process_text.register_async("async_plugin") async def async_handler(text: str) -> str: await asyncio.sleep(0.1) return text.lower() result = await process_text.apply_async("HELLO") """ def __init__( self, name: str, description: str = "", default_value: T | None = None, ) -> None: """ Initialisiert den Hook. Args: name: Eindeutiger Hook-Name description: Beschreibung default_value: Standardwert wenn keine Handler """ self._name = name self._description = description self._default_value = default_value self._handlers: list[HookHandler] = [] @property def name(self) -> str: """Hook-Name.""" return self._name @property def description(self) -> str: """Hook-Beschreibung.""" return self._description @property def handler_count(self) -> int: """Anzahl registrierter Handler.""" return len(self._handlers) def register( self, plugin_name: str, priority: HookPriority = HookPriority.NORMAL, once: bool = False, ) -> Callable[[HookCallback], HookCallback]: """ Decorator zum Registrieren eines Handlers. Args: plugin_name: Name des Plugins priority: Ausführungspriorität once: Nur einmal ausführen Returns: Decorator-Funktion """ def decorator(callback: HookCallback) -> HookCallback: handler = HookHandler( callback=callback, plugin_name=plugin_name, priority=priority, is_async=asyncio.iscoroutinefunction(callback), once=once, ) self._handlers.append(handler) self._handlers.sort() return callback return decorator def register_async( self, plugin_name: str, priority: HookPriority = HookPriority.NORMAL, once: bool = False, ) -> Callable[[AsyncHookCallback], AsyncHookCallback]: """Decorator für async Handler.""" return self.register(plugin_name, priority, once) def add_handler( self, callback: HookCallback, plugin_name: str, priority: HookPriority = HookPriority.NORMAL, once: bool = False, ) -> None: """Fügt einen Handler direkt hinzu.""" handler = HookHandler( callback=callback, plugin_name=plugin_name, priority=priority, is_async=asyncio.iscoroutinefunction(callback), once=once, ) self._handlers.append(handler) self._handlers.sort() def remove_handler(self, plugin_name: str) -> int: """ Entfernt alle Handler eines Plugins. Args: plugin_name: Plugin-Name Returns: Anzahl entfernter Handler """ before = len(self._handlers) self._handlers = [h for h in self._handlers if h.plugin_name != plugin_name] return before - len(self._handlers) def apply(self, value: T, *args: Any, **kwargs: Any) -> HookResult[T]: """ Wendet den Hook auf einen Wert an. Args: value: Der zu verarbeitende Wert *args: Zusätzliche Argumente **kwargs: Zusätzliche Keyword-Argumente Returns: HookResult mit dem verarbeiteten Wert """ current_value = value modified = False errors: list[str] = [] source = "" to_remove: list[HookHandler] = [] for handler in self._handlers: if not handler.enabled: continue try: if handler.is_async: # Async-Handler synchron ausführen (wenn möglich) try: loop = asyncio.get_event_loop() result = loop.run_until_complete( handler.callback(current_value, *args, **kwargs) ) except RuntimeError: # Kein Event-Loop - überspringen errors.append( f"{handler.plugin_name}: Async-Handler ohne Event-Loop" ) continue else: result = handler.callback(current_value, *args, **kwargs) handler.call_count += 1 if result is not None: current_value = result modified = True source = handler.plugin_name if handler.once: to_remove.append(handler) except Exception as e: errors.append(f"{handler.plugin_name}: {str(e)}") # Once-Handler entfernen for handler in to_remove: self._handlers.remove(handler) return HookResult( value=current_value, modified=modified, source=source, errors=errors, ) async def apply_async( self, value: T, *args: Any, **kwargs: Any, ) -> HookResult[T]: """ Wendet den Hook asynchron an. Args: value: Der zu verarbeitende Wert *args: Zusätzliche Argumente **kwargs: Zusätzliche Keyword-Argumente Returns: HookResult mit dem verarbeiteten Wert """ current_value = value modified = False errors: list[str] = [] source = "" to_remove: list[HookHandler] = [] for handler in self._handlers: if not handler.enabled: continue try: if handler.is_async: result = await handler.callback(current_value, *args, **kwargs) else: result = handler.callback(current_value, *args, **kwargs) handler.call_count += 1 if result is not None: current_value = result modified = True source = handler.plugin_name if handler.once: to_remove.append(handler) except Exception as e: errors.append(f"{handler.plugin_name}: {str(e)}") # Once-Handler entfernen for handler in to_remove: self._handlers.remove(handler) return HookResult( value=current_value, modified=modified, source=source, errors=errors, ) def trigger(self, *args: Any, **kwargs: Any) -> list[Any]: """ Triggert den Hook als Action (ohne Rückgabewert-Verkettung). Args: *args: Argumente für Handler **kwargs: Keyword-Argumente Returns: Liste der Rückgabewerte """ results = [] for handler in self._handlers: if not handler.enabled: continue try: if handler.is_async: continue # Async-Handler überspringen result = handler.callback(*args, **kwargs) handler.call_count += 1 results.append(result) if handler.once: self._handlers.remove(handler) except Exception: pass return results async def trigger_async(self, *args: Any, **kwargs: Any) -> list[Any]: """Triggert den Hook asynchron.""" results = [] for handler in list(self._handlers): if not handler.enabled: continue try: if handler.is_async: result = await handler.callback(*args, **kwargs) else: result = handler.callback(*args, **kwargs) handler.call_count += 1 results.append(result) if handler.once: self._handlers.remove(handler) except Exception: pass return results def clear(self) -> int: """ Entfernt alle Handler. Returns: Anzahl entfernter Handler """ count = len(self._handlers) self._handlers.clear() return count def get_handlers(self) -> list[HookHandler]: """Gibt alle Handler zurück.""" return self._handlers.copy() def __repr__(self) -> str: return f"PluginHook({self._name!r}, handlers={len(self._handlers)})"