Trixy7/trixy_core/service/circuit_breaker/breaker.py

# -*- coding: utf-8 -*-
"""
Circuit-Breaker-Implementierung.

Schützt Services vor kaskadierenden Fehlern durch
automatisches Unterbrechen fehlgeschlagener Verbindungen.
"""

from __future__ import annotations

import asyncio
import functools
import threading
import time
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Any, Callable, ParamSpec, Type, TypeVar

from trixy_core.service.circuit_breaker.states import (
    CircuitEvent,
    CircuitState,
    CircuitStateMachine,
    StateTransition,
)


P = ParamSpec("P")
R = TypeVar("R")


class CircuitBreakerError(Exception):
    """
    Basis-Exception für Circuit-Breaker-Fehler.
    """

    def __init__(
        self,
        message: str,
        breaker_name: str = "",
        state: CircuitState | None = None,
    ) -> None:
        """
        Initialisiert den Fehler.

        Args:
            message: Fehlermeldung.
            breaker_name: Name des Circuit-Breakers.
            state: Aktueller Zustand.
        """
        super().__init__(message)
        self.breaker_name = breaker_name
        self.state = state


class CircuitOpenError(CircuitBreakerError):
    """
    Exception wenn der Circuit offen ist.

    Wird geworfen wenn ein Request abgelehnt wird,
    weil der Circuit-Breaker offen ist.
    """

    def __init__(
        self,
        breaker_name: str,
        retry_after: float = 0.0,
    ) -> None:
        """
        Initialisiert den Fehler.

        Args:
            breaker_name: Name des Circuit-Breakers.
            retry_after: Zeit in Sekunden bis zum nächsten Versuch.
        """
        super().__init__(
            f"Circuit '{breaker_name}' ist offen",
            breaker_name=breaker_name,
            state=CircuitState.OPEN,
        )
        self.retry_after = retry_after


@dataclass
class CircuitBreakerConfig:
    """
    Konfiguration für einen Circuit-Breaker.
    """

    failure_threshold: int = 5
    """Anzahl Fehler bevor Circuit öffnet."""

    success_threshold: int = 3
    """Anzahl Erfolge in HALF_OPEN bevor Circuit schließt."""

    timeout: float = 30.0
    """Zeit in Sekunden bevor OPEN → HALF_OPEN."""

    half_open_max_calls: int = 3
    """Maximale Requests in HALF_OPEN."""

    failure_rate_threshold: float = 0.5
    """Fehlerrate (0.0-1.0) bevor Circuit öffnet."""

    min_calls_before_rate_check: int = 10
    """Mindest-Calls bevor Fehlerrate geprüft wird."""

    exceptions: tuple[Type[Exception], ...] = (Exception,)
    """Exceptions die als Fehler gezählt werden."""

    exclude_exceptions: tuple[Type[Exception], ...] = ()
    """Exceptions die nicht als Fehler gezählt werden."""

    on_open: Callable[[str], None] | None = None
    """Callback wenn Circuit öffnet."""

    on_close: Callable[[str], None] | None = None
    """Callback wenn Circuit schließt."""

    on_half_open: Callable[[str], None] | None = None
    """Callback wenn Circuit halb-offen wird."""


@dataclass
class CircuitBreakerStats:
    """
    Statistiken für einen Circuit-Breaker.
    """

    total_calls: int = 0
    """Gesamtzahl der Aufrufe."""

    successful_calls: int = 0
    """Erfolgreiche Aufrufe."""

    failed_calls: int = 0
    """Fehlgeschlagene Aufrufe."""

    rejected_calls: int = 0
    """Abgelehnte Aufrufe (Circuit offen)."""

    times_opened: int = 0
    """Wie oft der Circuit geöffnet wurde."""

    last_failure: datetime | None = None
    """Zeitpunkt des letzten Fehlers."""

    last_success: datetime | None = None
    """Zeitpunkt des letzten Erfolgs."""

    consecutive_failures: int = 0
    """Aktuelle Anzahl aufeinanderfolgender Fehler."""

    consecutive_successes: int = 0
    """Aktuelle Anzahl aufeinanderfolgender Erfolge."""

    @property
    def failure_rate(self) -> float:
        """Aktuelle Fehlerrate."""
        total = self.successful_calls + self.failed_calls
        if total == 0:
            return 0.0
        return self.failed_calls / total

    def record_success(self) -> None:
        """Zeichnet einen Erfolg auf."""
        self.total_calls += 1
        self.successful_calls += 1
        self.last_success = datetime.now()
        self.consecutive_successes += 1
        self.consecutive_failures = 0

    def record_failure(self) -> None:
        """Zeichnet einen Fehler auf."""
        self.total_calls += 1
        self.failed_calls += 1
        self.last_failure = datetime.now()
        self.consecutive_failures += 1
        self.consecutive_successes = 0

    def record_rejection(self) -> None:
        """Zeichnet eine Ablehnung auf."""
        self.rejected_calls += 1

    def reset(self) -> None:
        """Setzt die Statistiken zurück."""
        self.total_calls = 0
        self.successful_calls = 0
        self.failed_calls = 0
        self.rejected_calls = 0
        self.consecutive_failures = 0
        self.consecutive_successes = 0

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert in ein Dictionary."""
        return {
            "total_calls": self.total_calls,
            "successful_calls": self.successful_calls,
            "failed_calls": self.failed_calls,
            "rejected_calls": self.rejected_calls,
            "failure_rate": self.failure_rate,
            "times_opened": self.times_opened,
            "consecutive_failures": self.consecutive_failures,
            "consecutive_successes": self.consecutive_successes,
        }


class CircuitBreaker:
    """
    Circuit-Breaker für Fault Tolerance.

    Schützt vor kaskadierenden Fehlern durch automatisches
    Unterbrechen bei zu vielen Fehlern.

    Example:
        breaker = CircuitBreaker("database")

        try:
            with breaker:
                result = database.query("SELECT * FROM users")
        except CircuitOpenError:
            # Circuit ist offen, verwende Fallback
            result = cache.get("users")

        # Oder als Decorator
        @breaker
        def database_query():
            return database.query("SELECT * FROM users")
    """

    def __init__(
        self,
        name: str,
        config: CircuitBreakerConfig | None = None,
    ) -> None:
        """
        Initialisiert den Circuit-Breaker.

        Args:
            name: Name des Circuit-Breakers.
            config: Konfiguration.
        """
        self._name = name
        self._config = config or CircuitBreakerConfig()
        self._state_machine = CircuitStateMachine()
        self._stats = CircuitBreakerStats()
        self._lock = threading.Lock()

        self._opened_at: datetime | None = None
        self._half_open_calls = 0

    @property
    def name(self) -> str:
        """Name des Circuit-Breakers."""
        return self._name

    @property
    def state(self) -> CircuitState:
        """Aktueller Zustand."""
        return self._state_machine.state

    @property
    def stats(self) -> CircuitBreakerStats:
        """Statistiken."""
        return self._stats

    @property
    def config(self) -> CircuitBreakerConfig:
        """Konfiguration."""
        return self._config

    @property
    def is_closed(self) -> bool:
        """Prüft, ob der Circuit geschlossen ist."""
        return self.state == CircuitState.CLOSED

    @property
    def is_open(self) -> bool:
        """Prüft, ob der Circuit offen ist."""
        return self.state == CircuitState.OPEN

    @property
    def is_half_open(self) -> bool:
        """Prüft, ob der Circuit halb-offen ist."""
        return self.state == CircuitState.HALF_OPEN

    def _should_open(self) -> bool:
        """Prüft, ob der Circuit öffnen sollte."""
        # Failure-Threshold erreicht?
        if self._stats.consecutive_failures >= self._config.failure_threshold:
            return True

        # Failure-Rate zu hoch?
        if (
            self._stats.total_calls >= self._config.min_calls_before_rate_check
            and self._stats.failure_rate >= self._config.failure_rate_threshold
        ):
            return True

        return False

    def _should_close(self) -> bool:
        """Prüft, ob der Circuit schließen sollte (in HALF_OPEN)."""
        return self._stats.consecutive_successes >= self._config.success_threshold

    def _check_timeout(self) -> bool:
        """Prüft, ob der Open-Timeout abgelaufen ist."""
        if self._opened_at is None:
            return False

        elapsed = (datetime.now() - self._opened_at).total_seconds()
        return elapsed >= self._config.timeout

    def _transition_to(self, state: CircuitState, reason: str = "") -> None:
        """Führt einen Zustandsübergang durch."""
        old_state = self.state

        if state == CircuitState.OPEN:
            self._state_machine.force_state(CircuitState.OPEN, reason)
            self._opened_at = datetime.now()
            self._stats.times_opened += 1
            if self._config.on_open:
                self._config.on_open(self._name)

        elif state == CircuitState.HALF_OPEN:
            self._state_machine.force_state(CircuitState.HALF_OPEN, reason)
            self._half_open_calls = 0
            if self._config.on_half_open:
                self._config.on_half_open(self._name)

        elif state == CircuitState.CLOSED:
            self._state_machine.force_state(CircuitState.CLOSED, reason)
            self._opened_at = None
            self._half_open_calls = 0
            self._stats.consecutive_failures = 0
            if self._config.on_close:
                self._config.on_close(self._name)

    def allow_request(self) -> bool:
        """
        Prüft, ob ein Request erlaubt ist.

        Returns:
            True wenn erlaubt.
        """
        with self._lock:
            if self.state == CircuitState.CLOSED:
                return True

            if self.state == CircuitState.OPEN:
                # Timeout prüfen
                if self._check_timeout():
                    self._transition_to(CircuitState.HALF_OPEN, "Timeout")
                    return True
                return False

            if self.state == CircuitState.HALF_OPEN:
                if self._half_open_calls < self._config.half_open_max_calls:
                    self._half_open_calls += 1
                    return True
                return False

            return False

    def record_success(self) -> None:
        """Zeichnet einen Erfolg auf."""
        with self._lock:
            self._stats.record_success()

            if self.state == CircuitState.HALF_OPEN:
                if self._should_close():
                    self._transition_to(CircuitState.CLOSED, "Erfolge erreicht")

    def record_failure(self, exception: Exception | None = None) -> None:
        """
        Zeichnet einen Fehler auf.

        Args:
            exception: Die aufgetretene Exception.
        """
        with self._lock:
            # Prüfen ob Exception als Fehler zählt
            if exception:
                if isinstance(exception, self._config.exclude_exceptions):
                    return
                if not isinstance(exception, self._config.exceptions):
                    return

            self._stats.record_failure()

            if self.state == CircuitState.CLOSED:
                if self._should_open():
                    self._transition_to(CircuitState.OPEN, "Fehler-Threshold")

            elif self.state == CircuitState.HALF_OPEN:
                self._transition_to(CircuitState.OPEN, "Fehler in Half-Open")

    def __enter__(self) -> "CircuitBreaker":
        """Context-Manager-Eintritt."""
        if not self.allow_request():
            retry_after = 0.0
            if self._opened_at:
                elapsed = (datetime.now() - self._opened_at).total_seconds()
                retry_after = max(0, self._config.timeout - elapsed)

            self._stats.record_rejection()
            raise CircuitOpenError(self._name, retry_after)

        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
        """Context-Manager-Austritt."""
        if exc_type is None:
            self.record_success()
        else:
            self.record_failure(exc_val)

        return False  # Exception nicht unterdrücken

    def __call__(self, func: Callable[P, R]) -> Callable[P, R]:
        """Decorator-Verwendung."""
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            with self:
                return func(*args, **kwargs)

        return wrapper

    def force_open(self) -> None:
        """Erzwingt das Öffnen des Circuits."""
        with self._lock:
            self._transition_to(CircuitState.OPEN, "Manuell")

    def force_close(self) -> None:
        """Erzwingt das Schließen des Circuits."""
        with self._lock:
            self._transition_to(CircuitState.CLOSED, "Manuell")

    def reset(self) -> None:
        """Setzt den Circuit-Breaker zurück."""
        with self._lock:
            self._transition_to(CircuitState.CLOSED, "Reset")
            self._stats.reset()
            self._state_machine.reset()


def circuit_breaker(
    name: str | None = None,
    failure_threshold: int = 5,
    success_threshold: int = 3,
    timeout: float = 30.0,
    exceptions: tuple[Type[Exception], ...] = (Exception,),
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """
    Decorator für Circuit-Breaker.

    Args:
        name: Name des Circuit-Breakers (Default: Funktionsname).
        failure_threshold: Anzahl Fehler vor Öffnen.
        success_threshold: Anzahl Erfolge vor Schließen.
        timeout: Timeout in Sekunden.
        exceptions: Exceptions die als Fehler zählen.

    Returns:
        Decorator-Funktion.

    Example:
        @circuit_breaker(failure_threshold=3)
        def external_api_call():
            return requests.get(url).json()
    """
    config = CircuitBreakerConfig(
        failure_threshold=failure_threshold,
        success_threshold=success_threshold,
        timeout=timeout,
        exceptions=exceptions,
    )

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        breaker_name = name or func.__qualname__
        breaker = CircuitBreaker(breaker_name, config)

        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            with breaker:
                return func(*args, **kwargs)

        # Zugriff auf Breaker
        wrapper.circuit_breaker = breaker  # type: ignore

        return wrapper

    return decorator


# Registry für globale Circuit-Breaker
_breakers: dict[str, CircuitBreaker] = {}


def get_circuit_breaker(
    name: str,
    config: CircuitBreakerConfig | None = None,
) -> CircuitBreaker:
    """
    Holt oder erstellt einen Circuit-Breaker.

    Args:
        name: Name des Circuit-Breakers.
        config: Optionale Konfiguration.

    Returns:
        CircuitBreaker-Instanz.
    """
    if name not in _breakers:
        _breakers[name] = CircuitBreaker(name, config)
    return _breakers[name]


def get_all_circuit_breakers() -> dict[str, CircuitBreaker]:
    """Gibt alle registrierten Circuit-Breaker zurück."""
    return dict(_breakers)