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