Trixy7/trixy_core/network/retry.py

# -*- coding: utf-8 -*-
"""
Network Retry-Mechanismen.

Bietet robuste Retry-Strategien für Netzwerkoperationen mit
exponential backoff, Jitter und konfigurierbaren Policies.
"""

from __future__ import annotations

import asyncio
import random
import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum, auto
from typing import Any, Awaitable, Callable, TypeVar, Generic

T = TypeVar("T")


class RetryStrategy(Enum):
    """Verfügbare Retry-Strategien."""
    EXPONENTIAL = auto()     # Exponentielles Backoff
    LINEAR = auto()          # Lineares Backoff
    CONSTANT = auto()        # Konstante Wartezeit
    FIBONACCI = auto()       # Fibonacci-Sequenz
    DECORRELATED = auto()    # Decorrelated Jitter (AWS-Stil)


class RetryableError(Exception):
    """Exception die einen Retry auslösen soll."""
    pass


class NonRetryableError(Exception):
    """Exception die keinen Retry auslösen soll."""
    pass


@dataclass
class RetryConfig:
    """Konfiguration für Retry-Verhalten."""
    max_attempts: int = 3
    strategy: RetryStrategy = RetryStrategy.EXPONENTIAL
    base_delay: float = 1.0           # Basis-Wartezeit in Sekunden
    max_delay: float = 60.0           # Maximale Wartezeit
    multiplier: float = 2.0           # Multiplikator für exponentielles Backoff
    jitter: float = 0.1               # Zufällige Variation (0.0-1.0)
    retryable_exceptions: tuple = (Exception,)  # Exceptions die retry auslösen
    non_retryable_exceptions: tuple = ()        # Exceptions die NICHT retry auslösen
    on_retry: Callable[[int, Exception, float], None] | None = None  # Callback bei Retry


@dataclass
class RetryStats:
    """Statistiken über Retry-Versuche."""
    total_attempts: int = 0
    successful_attempts: int = 0
    failed_attempts: int = 0
    total_delay: float = 0.0
    last_error: Exception | None = None
    last_attempt_at: datetime | None = None
    history: list[dict[str, Any]] = field(default_factory=list)


class RetryContext:
    """
    Kontext für einen Retry-Durchlauf.

    Enthält Informationen über den aktuellen Versuch
    und ermöglicht Manipulation des Retry-Verhaltens.
    """

    def __init__(
        self,
        attempt: int,
        max_attempts: int,
        elapsed_time: float = 0.0,
    ) -> None:
        self.attempt = attempt
        self.max_attempts = max_attempts
        self.elapsed_time = elapsed_time
        self.should_retry = True
        self.next_delay: float | None = None
        self.metadata: dict[str, Any] = {}

    @property
    def is_last_attempt(self) -> bool:
        """Prüft ob dies der letzte Versuch ist."""
        return self.attempt >= self.max_attempts

    @property
    def remaining_attempts(self) -> int:
        """Anzahl verbleibender Versuche."""
        return max(0, self.max_attempts - self.attempt)

    def abort(self) -> None:
        """Bricht Retry ab - keine weiteren Versuche."""
        self.should_retry = False

    def set_next_delay(self, delay: float) -> None:
        """Setzt benutzerdefinierte Verzögerung für nächsten Versuch."""
        self.next_delay = delay


class NetworkRetry(Generic[T]):
    """
    Retry-Handler für Netzwerkoperationen.

    Bietet verschiedene Backoff-Strategien und detaillierte
    Statistiken über Retry-Versuche.

    Beispiel:
        retry = NetworkRetry(
            config=RetryConfig(
                max_attempts=5,
                strategy=RetryStrategy.EXPONENTIAL,
                base_delay=1.0,
            )
        )

        result = await retry.execute(async_operation)
    """

    def __init__(
        self,
        config: RetryConfig | None = None,
        logger: logging.Logger | None = None,
    ) -> None:
        """
        Initialisiert den Retry-Handler.

        Args:
            config: Retry-Konfiguration
            logger: Optional: Logger für Retry-Logs
        """
        self.config = config or RetryConfig()
        self.logger = logger or logging.getLogger(__name__)
        self.stats = RetryStats()
        self._fibonacci_cache: list[int] = [0, 1]

    def _get_fibonacci(self, n: int) -> int:
        """Berechnet Fibonacci-Zahl (gecached)."""
        while len(self._fibonacci_cache) <= n:
            self._fibonacci_cache.append(
                self._fibonacci_cache[-1] + self._fibonacci_cache[-2]
            )
        return self._fibonacci_cache[n]

    def _calculate_delay(self, attempt: int, context: RetryContext) -> float:
        """Berechnet Wartezeit basierend auf Strategie."""
        # Benutzerdefinierte Verzögerung?
        if context.next_delay is not None:
            base_delay = context.next_delay
        else:
            strategy = self.config.strategy
            base = self.config.base_delay

            if strategy == RetryStrategy.EXPONENTIAL:
                base_delay = base * (self.config.multiplier ** (attempt - 1))

            elif strategy == RetryStrategy.LINEAR:
                base_delay = base * attempt

            elif strategy == RetryStrategy.CONSTANT:
                base_delay = base

            elif strategy == RetryStrategy.FIBONACCI:
                base_delay = base * self._get_fibonacci(attempt)

            elif strategy == RetryStrategy.DECORRELATED:
                # AWS-Stil decorrelated Jitter
                prev_delay = context.metadata.get("prev_delay", base)
                base_delay = random.uniform(base, prev_delay * 3)
                context.metadata["prev_delay"] = base_delay

            else:
                base_delay = base

        # Maximale Verzögerung
        base_delay = min(base_delay, self.config.max_delay)

        # Jitter hinzufügen
        if self.config.jitter > 0:
            jitter_range = base_delay * self.config.jitter
            jitter = random.uniform(-jitter_range, jitter_range)
            base_delay = max(0, base_delay + jitter)

        return base_delay

    def _should_retry(
        self,
        exception: Exception,
        context: RetryContext,
    ) -> bool:
        """Prüft ob bei dieser Exception ein Retry stattfinden soll."""
        if not context.should_retry:
            return False

        if context.is_last_attempt:
            return False

        # Non-retryable Exceptions
        if isinstance(exception, self.config.non_retryable_exceptions):
            return False

        if isinstance(exception, NonRetryableError):
            return False

        # Retryable Exceptions
        if isinstance(exception, RetryableError):
            return True

        if isinstance(exception, self.config.retryable_exceptions):
            return True

        return False

    async def execute(
        self,
        operation: Callable[[], Awaitable[T]],
        on_attempt: Callable[[RetryContext], None] | None = None,
    ) -> T:
        """
        Führt Operation mit Retry aus.

        Args:
            operation: Async-Funktion die ausgeführt werden soll
            on_attempt: Optional: Callback vor jedem Versuch

        Returns:
            Ergebnis der Operation

        Raises:
            Exception: Letzte Exception nach allen Versuchen
        """
        start_time = datetime.now()
        last_exception: Exception | None = None

        for attempt in range(1, self.config.max_attempts + 1):
            elapsed = (datetime.now() - start_time).total_seconds()
            context = RetryContext(
                attempt=attempt,
                max_attempts=self.config.max_attempts,
                elapsed_time=elapsed,
            )

            # Callback vor Versuch
            if on_attempt:
                on_attempt(context)

            # Statistik aktualisieren
            self.stats.total_attempts += 1
            self.stats.last_attempt_at = datetime.now()

            try:
                result = await operation()
                self.stats.successful_attempts += 1
                self.stats.history.append({
                    "attempt": attempt,
                    "success": True,
                    "timestamp": datetime.now().isoformat(),
                })
                return result

            except Exception as e:
                last_exception = e
                self.stats.last_error = e
                self.stats.history.append({
                    "attempt": attempt,
                    "success": False,
                    "error": str(e),
                    "error_type": type(e).__name__,
                    "timestamp": datetime.now().isoformat(),
                })

                if not self._should_retry(e, context):
                    self.stats.failed_attempts += 1
                    self.logger.debug(
                        f"Retry abgebrochen nach Versuch {attempt}: {e}"
                    )
                    raise

                # Verzögerung berechnen
                delay = self._calculate_delay(attempt, context)
                self.stats.total_delay += delay

                self.logger.debug(
                    f"Versuch {attempt}/{self.config.max_attempts} fehlgeschlagen, "
                    f"nächster Versuch in {delay:.2f}s: {e}"
                )

                # Callback bei Retry
                if self.config.on_retry:
                    self.config.on_retry(attempt, e, delay)

                # Warten
                await asyncio.sleep(delay)

        self.stats.failed_attempts += 1
        if last_exception:
            raise last_exception
        raise RuntimeError("Retry fehlgeschlagen ohne Exception")

    def reset_stats(self) -> None:
        """Setzt Statistiken zurück."""
        self.stats = RetryStats()


def with_retry(
    max_attempts: int = 3,
    strategy: RetryStrategy = RetryStrategy.EXPONENTIAL,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter: float = 0.1,
    retryable_exceptions: tuple = (Exception,),
    non_retryable_exceptions: tuple = (),
) -> Callable:
    """
    Decorator für automatisches Retry von Async-Funktionen.

    Args:
        max_attempts: Maximale Anzahl Versuche
        strategy: Backoff-Strategie
        base_delay: Basis-Wartezeit
        max_delay: Maximale Wartezeit
        jitter: Zufällige Variation
        retryable_exceptions: Exceptions die Retry auslösen
        non_retryable_exceptions: Exceptions ohne Retry

    Beispiel:
        @with_retry(max_attempts=5, base_delay=2.0)
        async def fetch_data():
            return await client.get("/api/data")
    """
    config = RetryConfig(
        max_attempts=max_attempts,
        strategy=strategy,
        base_delay=base_delay,
        max_delay=max_delay,
        jitter=jitter,
        retryable_exceptions=retryable_exceptions,
        non_retryable_exceptions=non_retryable_exceptions,
    )

    def decorator(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
        async def wrapper(*args: Any, **kwargs: Any) -> T:
            retry = NetworkRetry[T](config=config)
            return await retry.execute(lambda: func(*args, **kwargs))

        wrapper.__name__ = func.__name__
        wrapper.__doc__ = func.__doc__
        return wrapper

    return decorator


class RetryPolicy:
    """
    Wiederverwendbare Retry-Policy.

    Erlaubt Definition von Retry-Verhalten das auf mehrere
    Operationen angewendet werden kann.
    """

    def __init__(
        self,
        name: str,
        config: RetryConfig,
    ) -> None:
        """
        Initialisiert die Policy.

        Args:
            name: Name der Policy
            config: Retry-Konfiguration
        """
        self.name = name
        self.config = config
        self._stats: dict[str, RetryStats] = {}

    async def execute(
        self,
        operation: Callable[[], Awaitable[T]],
        operation_name: str = "default",
    ) -> T:
        """
        Führt Operation mit dieser Policy aus.

        Args:
            operation: Auszuführende Operation
            operation_name: Name für Statistik-Tracking

        Returns:
            Ergebnis der Operation
        """
        retry = NetworkRetry[T](config=self.config)
        result = await retry.execute(operation)

        # Statistik speichern
        self._stats[operation_name] = retry.stats

        return result

    def get_stats(self, operation_name: str | None = None) -> RetryStats | dict[str, RetryStats]:
        """Holt Statistiken für Operation(en)."""
        if operation_name:
            return self._stats.get(operation_name, RetryStats())
        return self._stats.copy()


# Vordefinierte Policies
AGGRESSIVE_RETRY = RetryConfig(
    max_attempts=10,
    strategy=RetryStrategy.EXPONENTIAL,
    base_delay=0.5,
    max_delay=30.0,
    jitter=0.2,
)

CONSERVATIVE_RETRY = RetryConfig(
    max_attempts=3,
    strategy=RetryStrategy.LINEAR,
    base_delay=5.0,
    max_delay=60.0,
    jitter=0.1,
)

FAST_FAIL_RETRY = RetryConfig(
    max_attempts=2,
    strategy=RetryStrategy.CONSTANT,
    base_delay=1.0,
    max_delay=1.0,
    jitter=0.0,
)