Trixy7/trixy_core/utils/logging/context.py

# -*- 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)