# -*- coding: utf-8 -*- """ Logging-Kontext für Korrelations-IDs und kontextuelle Metadaten. Ermöglicht die Propagation von Kontext-Informationen durch die gesamte Anwendung für besseres Tracing und Debugging. """ from __future__ import annotations import contextvars import uuid from contextlib import contextmanager from dataclasses import dataclass, field from datetime import datetime from typing import Any, Generator, Iterator # Kontextvariablen für Thread-/Task-lokale Speicherung _correlation_id: contextvars.ContextVar[str | None] = contextvars.ContextVar( "correlation_id", default=None ) _log_context: contextvars.ContextVar[dict[str, Any]] = contextvars.ContextVar( "log_context", default={} ) _context_stack: contextvars.ContextVar[list[dict[str, Any]]] = contextvars.ContextVar( "context_stack", default=[] ) def get_correlation_id() -> str: """ Gibt die aktuelle Korrelations-ID zurück. Generiert automatisch eine neue ID, falls keine existiert. Returns: Die aktuelle Korrelations-ID. """ cid = _correlation_id.get() if cid is None: cid = str(uuid.uuid4()) _correlation_id.set(cid) return cid def set_correlation_id(correlation_id: str | None) -> str | None: """ Setzt die Korrelations-ID für den aktuellen Kontext. Args: correlation_id: Die zu setzende ID oder None zum Zurücksetzen. Returns: Die vorherige Korrelations-ID. """ previous = _correlation_id.get() _correlation_id.set(correlation_id) return previous def generate_correlation_id() -> str: """ Generiert eine neue Korrelations-ID und setzt sie im Kontext. Returns: Die neu generierte Korrelations-ID. """ new_id = str(uuid.uuid4()) _correlation_id.set(new_id) return new_id def get_context() -> dict[str, Any]: """ Gibt den aktuellen Log-Kontext zurück. Returns: Dictionary mit kontextuellen Metadaten. """ return dict(_log_context.get()) def set_context(**kwargs: Any) -> None: """ Setzt Werte im aktuellen Log-Kontext. Args: **kwargs: Schlüssel-Wert-Paare für den Kontext. """ current = dict(_log_context.get()) current.update(kwargs) _log_context.set(current) def clear_context() -> None: """Löscht den gesamten Log-Kontext.""" _log_context.set({}) _correlation_id.set(None) @contextmanager def log_context(**kwargs: Any) -> Generator[dict[str, Any], None, None]: """ Context-Manager für temporäre Log-Kontext-Erweiterung. Fügt die angegebenen Werte zum Kontext hinzu und entfernt sie beim Verlassen des Blocks automatisch. Args: **kwargs: Temporäre Kontext-Werte. Yields: Der aktuelle kombinierte Kontext. Example: with log_context(user_id="123", action="login"): logger.info("Benutzer meldet sich an") """ previous = dict(_log_context.get()) new_context = {**previous, **kwargs} _log_context.set(new_context) try: yield new_context finally: _log_context.set(previous) @contextmanager def correlation_scope(correlation_id: str | None = None) -> Generator[str, None, None]: """ Context-Manager für Korrelations-ID-Scope. Args: correlation_id: Optionale ID. Falls None, wird eine neue generiert. Yields: Die aktive Korrelations-ID. Example: with correlation_scope() as cid: logger.info("Request gestartet") # Alle Logs haben dieselbe correlation_id """ previous = _correlation_id.get() new_id = correlation_id if correlation_id else str(uuid.uuid4()) _correlation_id.set(new_id) try: yield new_id finally: _correlation_id.set(previous) @dataclass class LogContext: """ Strukturierter Log-Kontext mit allen relevanten Metadaten. Sammelt automatisch Korrelations-ID, Zeitstempel und benutzerdefinierte Attribute für strukturiertes Logging. Attributes: correlation_id: Eindeutige ID für Request-Tracing. timestamp: Zeitpunkt der Kontext-Erstellung. service_name: Name des aktuellen Services. environment: Umgebung (production, staging, development). extra: Zusätzliche benutzerdefinierte Felder. """ correlation_id: str = field(default_factory=get_correlation_id) timestamp: datetime = field(default_factory=datetime.now) service_name: str = "" environment: str = "" extra: dict[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: """Initialisiert den Kontext mit gespeicherten Werten.""" # Übernimm Werte aus dem globalen Kontext ctx = get_context() if not self.service_name and "service_name" in ctx: self.service_name = ctx["service_name"] if not self.environment and "environment" in ctx: self.environment = ctx["environment"] def to_dict(self) -> dict[str, Any]: """ Konvertiert den Kontext in ein Dictionary. Returns: Dictionary mit allen Kontext-Feldern. """ result: dict[str, Any] = { "correlation_id": self.correlation_id, "timestamp": self.timestamp.isoformat(), } if self.service_name: result["service_name"] = self.service_name if self.environment: result["environment"] = self.environment if self.extra: result.update(self.extra) return result def with_extra(self, **kwargs: Any) -> "LogContext": """ Erstellt eine Kopie mit zusätzlichen Feldern. Args: **kwargs: Zusätzliche Felder. Returns: Neue LogContext-Instanz mit erweiterten Feldern. """ new_extra = {**self.extra, **kwargs} return LogContext( correlation_id=self.correlation_id, timestamp=self.timestamp, service_name=self.service_name, environment=self.environment, extra=new_extra, ) @classmethod def from_headers(cls, headers: dict[str, str]) -> "LogContext": """ Erstellt einen Kontext aus HTTP-Headers. Nützlich für die Propagation von Kontext-Informationen zwischen Services. Args: headers: HTTP-Headers mit Kontext-Informationen. Returns: LogContext mit Werten aus den Headers. """ correlation_id = headers.get( "X-Correlation-ID", headers.get("X-Request-ID", str(uuid.uuid4())) ) return cls( correlation_id=correlation_id, service_name=headers.get("X-Service-Name", ""), environment=headers.get("X-Environment", ""), ) def to_headers(self) -> dict[str, str]: """ Konvertiert den Kontext in HTTP-Headers. Returns: Dictionary mit Header-Namen und -Werten. """ headers = { "X-Correlation-ID": self.correlation_id, } if self.service_name: headers["X-Service-Name"] = self.service_name if self.environment: headers["X-Environment"] = self.environment return headers class ContextStack: """ Stack-basierte Kontext-Verwaltung für verschachtelte Operationen. Ermöglicht das Pushen und Poppen von Kontext-Layern für hierarchische Logging-Szenarien. """ def __init__(self) -> None: """Initialisiert einen leeren Kontext-Stack.""" self._stack: list[dict[str, Any]] = [] def push(self, **kwargs: Any) -> None: """ Fügt einen neuen Kontext-Layer zum Stack hinzu. Args: **kwargs: Kontext-Werte für den neuen Layer. """ current = dict(_log_context.get()) self._stack.append(current) new_context = {**current, **kwargs} _log_context.set(new_context) def pop(self) -> dict[str, Any] | None: """ Entfernt den obersten Kontext-Layer vom Stack. Returns: Der entfernte Kontext oder None, falls Stack leer. """ if self._stack: previous = self._stack.pop() _log_context.set(previous) return previous return None def peek(self) -> dict[str, Any]: """ Gibt den aktuellen Kontext zurück, ohne ihn zu entfernen. Returns: Der aktuelle kombinierte Kontext. """ return dict(_log_context.get()) @property def depth(self) -> int: """Gibt die aktuelle Stack-Tiefe zurück.""" return len(self._stack) @contextmanager def scope(self, **kwargs: Any) -> Iterator[dict[str, Any]]: """ Context-Manager für automatisches Push/Pop. Args: **kwargs: Kontext-Werte für den Scope. Yields: Der aktuelle Kontext innerhalb des Scopes. """ self.push(**kwargs) try: yield self.peek() finally: self.pop() # Globale Stack-Instanz für einfache Verwendung _global_stack = ContextStack() def push_context(**kwargs: Any) -> None: """Fügt Werte zum globalen Kontext-Stack hinzu.""" _global_stack.push(**kwargs) def pop_context() -> dict[str, Any] | None: """Entfernt den obersten Layer vom globalen Kontext-Stack.""" return _global_stack.pop() def context_scope(**kwargs: Any) -> Iterator[dict[str, Any]]: """Context-Manager für den globalen Kontext-Stack.""" return _global_stack.scope(**kwargs)