# -*- coding: utf-8 -*- """ Logging-Middleware für Events. Protokolliert Event-Verarbeitung für Debugging und Audit. """ from __future__ import annotations import json import logging from datetime import datetime from typing import TYPE_CHECKING, Any, Callable from trixy_core.events.middleware.base import ( EventMiddleware, MiddlewareContext, MiddlewareResult, ) if TYPE_CHECKING: from trixy_core.events.event_data.base import EventData class LoggingMiddleware(EventMiddleware): """ Middleware für Event-Logging. Protokolliert Event-Namen, Daten und Verarbeitungszeiten. Example: middleware = LoggingMiddleware( logger=logging.getLogger("events"), log_data=True, log_metadata=True, ) pipeline.add(middleware) """ def __init__( self, name: str = "LoggingMiddleware", enabled: bool = True, logger: logging.Logger | None = None, log_level: int = logging.DEBUG, log_data: bool = False, log_metadata: bool = False, include_events: list[str] | None = None, exclude_events: list[str] | None = None, data_formatter: Callable[["EventData"], dict[str, Any]] | None = None, ) -> None: """ Initialisiert die Logging-Middleware. Args: name: Name der Middleware. enabled: Ob aktiv. logger: Logger-Instanz (Standard: neuer Logger). log_level: Log-Level für Events. log_data: Ob Event-Daten geloggt werden. log_metadata: Ob Metadaten geloggt werden. include_events: Events die geloggt werden (oder alle). exclude_events: Events die nicht geloggt werden. data_formatter: Optionaler Formatter für Event-Daten. """ super().__init__(name, enabled) self._logger = logger or logging.getLogger("trixy.events") self._log_level = log_level self._log_data = log_data self._log_metadata = log_metadata self._include_events = set(include_events) if include_events else None self._exclude_events = set(exclude_events) if exclude_events else set() self._data_formatter = data_formatter def should_process(self, event_name: str) -> bool: """Prüft, ob das Event geloggt werden soll.""" if event_name in self._exclude_events: return False if self._include_events is not None: return event_name in self._include_events return True def _format_event_data(self, event_data: "EventData") -> dict[str, Any]: """ Formatiert Event-Daten für das Logging. Args: event_data: Die Event-Daten. Returns: Dictionary-Repräsentation. """ if self._data_formatter: return self._data_formatter(event_data) # Standard-Formatierung result: dict[str, Any] = { "event_id": event_data.event_id, "source": event_data.source, "cancelled": event_data.cancelled, } if self._log_metadata: result["metadata"] = event_data.metadata return result async def before( self, context: MiddlewareContext, ) -> MiddlewareResult: """ Loggt den Event-Start. Args: context: Der Middleware-Kontext. Returns: Ergebnis (immer CONTINUE). """ log_entry: dict[str, Any] = { "phase": "before", "event": context.event_name, "timestamp": context.started_at.isoformat(), } if self._log_data: log_entry["data"] = self._format_event_data(context.event_data) self._logger.log( self._log_level, f"Event START: {context.event_name}", extra={"event_log": log_entry}, ) return MiddlewareResult() async def after( self, context: MiddlewareContext, ) -> None: """ Loggt das Event-Ende. Args: context: Der Middleware-Kontext. """ log_entry: dict[str, Any] = { "phase": "after", "event": context.event_name, "duration_ms": context.elapsed_ms, "cancelled": context.event_data.cancelled, "skipped": context.skipped, "aborted": context.aborted, "error_count": len(context.errors), } if self._log_data: log_entry["data"] = self._format_event_data(context.event_data) level = self._log_level if context.errors: level = logging.WARNING self._logger.log( level, f"Event END: {context.event_name} ({context.elapsed_ms:.2f}ms)", extra={"event_log": log_entry}, ) async def on_error( self, context: MiddlewareContext, error: Exception, ) -> None: """ Loggt Event-Fehler. Args: context: Der Middleware-Kontext. error: Der aufgetretene Fehler. """ log_entry: dict[str, Any] = { "phase": "error", "event": context.event_name, "error_type": type(error).__name__, "error_message": str(error), "duration_ms": context.elapsed_ms, } self._logger.error( f"Event ERROR: {context.event_name} - {error}", extra={"event_log": log_entry}, exc_info=True, ) class StructuredLoggingMiddleware(EventMiddleware): """ Middleware für strukturiertes JSON-Logging. Gibt Event-Informationen als JSON aus für Log-Aggregation. Example: middleware = StructuredLoggingMiddleware( output=sys.stdout.write, include_stack_trace=True, ) """ def __init__( self, name: str = "StructuredLoggingMiddleware", enabled: bool = True, output: Callable[[str], Any] | None = None, include_data: bool = True, include_stack_trace: bool = False, pretty_print: bool = False, ) -> None: """ Initialisiert die Middleware. Args: name: Name der Middleware. enabled: Ob aktiv. output: Output-Funktion (Standard: print). include_data: Ob Event-Daten inkludiert werden. include_stack_trace: Ob Stack-Traces bei Fehlern inkludiert werden. pretty_print: Ob JSON formatiert ausgegeben wird. """ super().__init__(name, enabled) self._output = output or print self._include_data = include_data self._include_stack_trace = include_stack_trace self._pretty_print = pretty_print def _emit(self, log_entry: dict[str, Any]) -> None: """ Gibt einen Log-Eintrag aus. Args: log_entry: Der Log-Eintrag. """ if self._pretty_print: json_str = json.dumps(log_entry, indent=2, default=str) else: json_str = json.dumps(log_entry, default=str) self._output(json_str) async def before( self, context: MiddlewareContext, ) -> MiddlewareResult: """Loggt Event-Start als JSON.""" log_entry = { "@timestamp": datetime.now().isoformat(), "level": "INFO", "type": "event", "phase": "start", "event_name": context.event_name, "event_id": context.event_data.event_id, "source": context.event_data.source, } if self._include_data: log_entry["data"] = { k: v for k, v in context.event_data.__dict__.items() if not k.startswith("_") and k not in ("event_id", "source", "cancelled") } self._emit(log_entry) return MiddlewareResult() async def after( self, context: MiddlewareContext, ) -> None: """Loggt Event-Ende als JSON.""" log_entry = { "@timestamp": datetime.now().isoformat(), "level": "INFO" if not context.errors else "WARNING", "type": "event", "phase": "end", "event_name": context.event_name, "event_id": context.event_data.event_id, "duration_ms": context.elapsed_ms, "cancelled": context.event_data.cancelled, "skipped": context.skipped, "aborted": context.aborted, } if context.errors: log_entry["errors"] = [str(e) for e in context.errors] self._emit(log_entry) async def on_error( self, context: MiddlewareContext, error: Exception, ) -> None: """Loggt Fehler als JSON.""" import traceback log_entry = { "@timestamp": datetime.now().isoformat(), "level": "ERROR", "type": "event_error", "event_name": context.event_name, "event_id": context.event_data.event_id, "error_type": type(error).__name__, "error_message": str(error), "duration_ms": context.elapsed_ms, } if self._include_stack_trace: log_entry["stack_trace"] = traceback.format_exc() self._emit(log_entry)