# -*- coding: utf-8 -*- """ Basis-Middleware-Klassen für Events. Definiert die abstrakte Middleware und die Pipeline-Verwaltung. """ from __future__ import annotations import asyncio from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime from enum import Enum, auto from typing import TYPE_CHECKING, Any, Callable, Coroutine if TYPE_CHECKING: from trixy_core.events.event_data.base import EventData class MiddlewareAction(Enum): """ Aktionen die eine Middleware ausführen kann. """ CONTINUE = auto() """Normale Fortsetzung der Pipeline.""" SKIP_HANDLERS = auto() """Überspringt die Event-Handler, führt aber After-Hooks aus.""" ABORT = auto() """Bricht die gesamte Pipeline ab.""" RETRY = auto() """Wiederholt den aktuellen Schritt.""" @dataclass class MiddlewareResult: """ Ergebnis einer Middleware-Ausführung. """ action: MiddlewareAction = MiddlewareAction.CONTINUE """Aktion für die Pipeline.""" modified_data: "EventData | None" = None """Modifizierte Event-Daten (oder None wenn unverändert).""" error: Exception | None = None """Fehler, falls aufgetreten.""" metadata: dict[str, Any] = field(default_factory=dict) """Zusätzliche Metadaten.""" @property def should_continue(self) -> bool: """Prüft, ob die Pipeline fortgesetzt werden soll.""" return self.action == MiddlewareAction.CONTINUE @property def should_skip(self) -> bool: """Prüft, ob Handler übersprungen werden sollen.""" return self.action == MiddlewareAction.SKIP_HANDLERS @property def should_abort(self) -> bool: """Prüft, ob die Pipeline abgebrochen werden soll.""" return self.action == MiddlewareAction.ABORT @dataclass class MiddlewareContext: """ Kontext für die Middleware-Ausführung. Enthält Informationen über das aktuelle Event und ermöglicht die Kommunikation zwischen Middlewares. """ event_name: str """Name des Events.""" event_data: "EventData" """Die Event-Daten.""" started_at: datetime = field(default_factory=datetime.now) """Startzeitpunkt der Verarbeitung.""" properties: dict[str, Any] = field(default_factory=dict) """Geteilte Eigenschaften zwischen Middlewares.""" errors: list[Exception] = field(default_factory=list) """Gesammelte Fehler.""" skipped: bool = False """Ob Handler übersprungen wurden.""" aborted: bool = False """Ob die Pipeline abgebrochen wurde.""" def set_property(self, key: str, value: Any) -> None: """ Setzt eine Eigenschaft im Kontext. Args: key: Schlüssel. value: Wert. """ self.properties[key] = value def get_property(self, key: str, default: Any = None) -> Any: """ Gibt eine Eigenschaft zurück. Args: key: Schlüssel. default: Standardwert. Returns: Der Wert oder der Standardwert. """ return self.properties.get(key, default) @property def elapsed_ms(self) -> float: """Verstrichene Zeit in Millisekunden.""" return (datetime.now() - self.started_at).total_seconds() * 1000 class EventMiddleware(ABC): """ Abstrakte Basisklasse für Event-Middleware. Eine Middleware kann vor (before) und nach (after) der Event-Verarbeitung Code ausführen. Example: class MyMiddleware(EventMiddleware): async def before(self, context: MiddlewareContext) -> MiddlewareResult: print(f"Vor Event: {context.event_name}") return MiddlewareResult() async def after(self, context: MiddlewareContext) -> None: print(f"Nach Event: {context.event_name}") """ def __init__( self, name: str | None = None, enabled: bool = True, ) -> None: """ Initialisiert die Middleware. Args: name: Name der Middleware (Standard: Klassenname). enabled: Ob die Middleware aktiv ist. """ self._name = name or self.__class__.__name__ self._enabled = enabled @property def name(self) -> str: """Name der Middleware.""" return self._name @property def enabled(self) -> bool: """Ob die Middleware aktiv ist.""" return self._enabled @enabled.setter def enabled(self, value: bool) -> None: """Setzt den Aktivierungsstatus.""" self._enabled = value def should_process(self, event_name: str) -> bool: """ Prüft, ob das Event verarbeitet werden soll. Kann überschrieben werden, um Events zu filtern. Args: event_name: Name des Events. Returns: True wenn das Event verarbeitet werden soll. """ return True @abstractmethod async def before( self, context: MiddlewareContext, ) -> MiddlewareResult: """ Wird vor der Event-Verarbeitung aufgerufen. Args: context: Der Middleware-Kontext. Returns: Ergebnis mit der gewünschten Aktion. """ pass async def after( self, context: MiddlewareContext, ) -> None: """ Wird nach der Event-Verarbeitung aufgerufen. Args: context: Der Middleware-Kontext. """ pass async def on_error( self, context: MiddlewareContext, error: Exception, ) -> None: """ Wird bei einem Fehler aufgerufen. Args: context: Der Middleware-Kontext. error: Der aufgetretene Fehler. """ pass class PassThroughMiddleware(EventMiddleware): """ Einfache Middleware die nichts tut. Nützlich als Platzhalter oder Basis für einfache Implementierungen. """ async def before( self, context: MiddlewareContext, ) -> MiddlewareResult: """Gibt einfach CONTINUE zurück.""" return MiddlewareResult() class MiddlewarePipeline: """ Pipeline für die Ausführung von Middlewares. Verwaltet eine geordnete Liste von Middlewares und führt sie in der richtigen Reihenfolge aus. Example: pipeline = MiddlewarePipeline() pipeline.add(LoggingMiddleware()) pipeline.add(TimingMiddleware()) pipeline.add(ValidationMiddleware()) # Event verarbeiten async def process_event(event_name, event_data): context = MiddlewareContext(event_name, event_data) # Before-Phase result = await pipeline.run_before(context) if result.should_abort: return # Handler ausführen if not result.should_skip: await run_handlers(event_name, event_data) # After-Phase await pipeline.run_after(context) """ def __init__(self) -> None: """Initialisiert die Pipeline.""" self._middlewares: list[EventMiddleware] = [] @property def middlewares(self) -> list[EventMiddleware]: """Gibt die Middleware-Liste zurück.""" return list(self._middlewares) @property def count(self) -> int: """Anzahl der Middlewares.""" return len(self._middlewares) def add( self, middleware: EventMiddleware, position: int | None = None, ) -> None: """ Fügt eine Middleware hinzu. Args: middleware: Die Middleware. position: Position in der Liste (None = Ende). """ if position is None: self._middlewares.append(middleware) else: self._middlewares.insert(position, middleware) def remove(self, middleware: EventMiddleware) -> bool: """ Entfernt eine Middleware. Args: middleware: Die Middleware. Returns: True wenn entfernt. """ try: self._middlewares.remove(middleware) return True except ValueError: return False def remove_by_name(self, name: str) -> bool: """ Entfernt eine Middleware nach Name. Args: name: Name der Middleware. Returns: True wenn entfernt. """ for i, mw in enumerate(self._middlewares): if mw.name == name: self._middlewares.pop(i) return True return False def get(self, name: str) -> EventMiddleware | None: """ Gibt eine Middleware nach Name zurück. Args: name: Name der Middleware. Returns: Die Middleware oder None. """ for mw in self._middlewares: if mw.name == name: return mw return None def clear(self) -> None: """Entfernt alle Middlewares.""" self._middlewares.clear() async def run_before( self, context: MiddlewareContext, ) -> MiddlewareResult: """ Führt die Before-Phase aller Middlewares aus. Args: context: Der Middleware-Kontext. Returns: Das aggregierte Ergebnis. """ final_result = MiddlewareResult() for middleware in self._middlewares: if not middleware.enabled: continue if not middleware.should_process(context.event_name): continue try: result = await middleware.before(context) # Aktualisiere Event-Daten wenn modifiziert if result.modified_data is not None: context.event_data = result.modified_data # Prüfe Aktion if result.should_abort: context.aborted = True return result if result.should_skip: context.skipped = True final_result.action = MiddlewareAction.SKIP_HANDLERS except Exception as e: context.errors.append(e) await middleware.on_error(context, e) return final_result async def run_after( self, context: MiddlewareContext, ) -> None: """ Führt die After-Phase aller Middlewares aus. Middlewares werden in umgekehrter Reihenfolge ausgeführt. Args: context: Der Middleware-Kontext. """ # Umgekehrte Reihenfolge für After-Phase for middleware in reversed(self._middlewares): if not middleware.enabled: continue if not middleware.should_process(context.event_name): continue try: await middleware.after(context) except Exception as e: context.errors.append(e) await middleware.on_error(context, e) async def run_on_error( self, context: MiddlewareContext, error: Exception, ) -> None: """ Führt die Error-Handler aller Middlewares aus. Args: context: Der Middleware-Kontext. error: Der aufgetretene Fehler. """ for middleware in self._middlewares: if not middleware.enabled: continue try: await middleware.on_error(context, error) except Exception: # Fehler in Error-Handlern ignorieren pass def create_context( self, event_name: str, event_data: "EventData", ) -> MiddlewareContext: """ Erstellt einen neuen Middleware-Kontext. Args: event_name: Name des Events. event_data: Die Event-Daten. Returns: Neuer Kontext. """ return MiddlewareContext( event_name=event_name, event_data=event_data, )