Trixy7/trixy_core/service/circuit_breaker/states.py

# -*- coding: utf-8 -*-
"""
Circuit-Breaker-Zustände und Zustandsübergänge.

Definiert die Zustandsmaschine für den Circuit-Breaker.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto


class CircuitState(Enum):
    """
    Zustände des Circuit-Breakers.
    """

    CLOSED = auto()
    """
    Geschlossen - Normalbetrieb.
    Requests werden durchgelassen, Fehler werden gezählt.
    """

    OPEN = auto()
    """
    Offen - Alle Requests werden abgelehnt.
    Wartet auf Timeout bevor HALF_OPEN erreicht wird.
    """

    HALF_OPEN = auto()
    """
    Halb-Offen - Testphase.
    Begrenzte Anzahl Requests werden durchgelassen.
    Bei Erfolg → CLOSED, bei Fehler → OPEN.
    """


class CircuitEvent(Enum):
    """
    Ereignisse die Zustandsübergänge auslösen.
    """

    SUCCESS = auto()
    """Erfolgreicher Request."""

    FAILURE = auto()
    """Fehlgeschlagener Request."""

    TIMEOUT = auto()
    """Open-Timeout abgelaufen."""

    FORCE_OPEN = auto()
    """Manuelles Öffnen."""

    FORCE_CLOSE = auto()
    """Manuelles Schließen."""

    FORCE_RESET = auto()
    """Manueller Reset."""


@dataclass
class StateTransition:
    """
    Ein Zustandsübergang im Circuit-Breaker.
    """

    from_state: CircuitState
    """Ausgangszustand."""

    to_state: CircuitState
    """Zielzustand."""

    event: CircuitEvent
    """Auslösendes Ereignis."""

    timestamp: datetime = field(default_factory=datetime.now)
    """Zeitpunkt des Übergangs."""

    reason: str = ""
    """Grund für den Übergang."""

    def __str__(self) -> str:
        """String-Repräsentation."""
        return (
            f"{self.from_state.name} -> {self.to_state.name} "
            f"({self.event.name})"
        )


class CircuitStateMachine:
    """
    Zustandsmaschine für den Circuit-Breaker.

    Verwaltet die Zustandsübergänge basierend auf Ereignissen.
    """

    # Erlaubte Übergänge: (von, event) -> zu
    TRANSITIONS: dict[tuple[CircuitState, CircuitEvent], CircuitState] = {
        # CLOSED Zustand
        (CircuitState.CLOSED, CircuitEvent.SUCCESS): CircuitState.CLOSED,
        (CircuitState.CLOSED, CircuitEvent.FAILURE): CircuitState.CLOSED,  # Oder OPEN bei Threshold
        (CircuitState.CLOSED, CircuitEvent.FORCE_OPEN): CircuitState.OPEN,

        # OPEN Zustand
        (CircuitState.OPEN, CircuitEvent.TIMEOUT): CircuitState.HALF_OPEN,
        (CircuitState.OPEN, CircuitEvent.FORCE_CLOSE): CircuitState.CLOSED,
        (CircuitState.OPEN, CircuitEvent.FORCE_RESET): CircuitState.CLOSED,

        # HALF_OPEN Zustand
        (CircuitState.HALF_OPEN, CircuitEvent.SUCCESS): CircuitState.CLOSED,  # Oder HALF_OPEN
        (CircuitState.HALF_OPEN, CircuitEvent.FAILURE): CircuitState.OPEN,
        (CircuitState.HALF_OPEN, CircuitEvent.FORCE_OPEN): CircuitState.OPEN,
        (CircuitState.HALF_OPEN, CircuitEvent.FORCE_CLOSE): CircuitState.CLOSED,
    }

    def __init__(self, initial_state: CircuitState = CircuitState.CLOSED) -> None:
        """
        Initialisiert die Zustandsmaschine.

        Args:
            initial_state: Anfangszustand.
        """
        self._state = initial_state
        self._history: list[StateTransition] = []

    @property
    def state(self) -> CircuitState:
        """Gibt den aktuellen Zustand zurück."""
        return self._state

    @property
    def history(self) -> list[StateTransition]:
        """Gibt die Übergangshistorie zurück."""
        return list(self._history)

    def can_transition(
        self,
        event: CircuitEvent,
    ) -> tuple[bool, CircuitState | None]:
        """
        Prüft, ob ein Übergang möglich ist.

        Args:
            event: Das Ereignis.

        Returns:
            Tupel (möglich, Zielzustand).
        """
        key = (self._state, event)
        if key in self.TRANSITIONS:
            return True, self.TRANSITIONS[key]
        return False, None

    def transition(
        self,
        event: CircuitEvent,
        reason: str = "",
    ) -> StateTransition | None:
        """
        Führt einen Zustandsübergang durch.

        Args:
            event: Das auslösende Ereignis.
            reason: Optionaler Grund.

        Returns:
            StateTransition oder None wenn ungültig.
        """
        can_do, new_state = self.can_transition(event)

        if not can_do or new_state is None:
            return None

        old_state = self._state
        self._state = new_state

        transition = StateTransition(
            from_state=old_state,
            to_state=new_state,
            event=event,
            reason=reason,
        )

        self._history.append(transition)

        # Historie begrenzen
        if len(self._history) > 100:
            self._history = self._history[-50:]

        return transition

    def force_state(
        self,
        state: CircuitState,
        reason: str = "Forced",
    ) -> StateTransition:
        """
        Erzwingt einen bestimmten Zustand.

        Args:
            state: Zielzustand.
            reason: Grund.

        Returns:
            StateTransition.
        """
        old_state = self._state
        self._state = state

        transition = StateTransition(
            from_state=old_state,
            to_state=state,
            event=CircuitEvent.FORCE_RESET,
            reason=reason,
        )

        self._history.append(transition)
        return transition

    def reset(self) -> None:
        """Setzt die Zustandsmaschine zurück."""
        self._state = CircuitState.CLOSED
        self._history.clear()