Trixy7/trixy_core/utils/decorators/retry.py

# -*- coding: utf-8 -*-
"""
Retry-Decorators mit exponential Backoff und Jitter.

Bietet robuste Retry-Logik für fehleranfällige Operationen
wie Netzwerk-Requests oder Datenbank-Zugriffe.
"""

from __future__ import annotations

import asyncio
import functools
import logging
import random
import time
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Callable, ParamSpec, Sequence, Type, TypeVar


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


@dataclass
class RetryConfig:
    """
    Konfiguration für Retry-Verhalten.

    Definiert alle Parameter für die Retry-Logik inklusive
    Backoff-Strategie und Ausnahmebehandlung.
    """

    max_attempts: int = 3
    """Maximale Anzahl von Versuchen (inklusive dem ersten)."""

    base_delay: float = 1.0
    """Basis-Verzögerung in Sekunden."""

    max_delay: float = 60.0
    """Maximale Verzögerung in Sekunden."""

    exponential_base: float = 2.0
    """Basis für exponentielles Backoff (2.0 = Verdoppelung)."""

    jitter: float = 0.1
    """Jitter-Faktor (0.0 - 1.0). Fügt Zufälligkeit hinzu."""

    retry_on: tuple[Type[BaseException], ...] = (Exception,)
    """Ausnahmen, bei denen ein Retry erfolgen soll."""

    retry_if: Callable[[BaseException], bool] | None = None
    """Optionale Funktion zur Entscheidung, ob Retry erfolgen soll."""

    on_retry: Callable[[int, BaseException, float], None] | None = None
    """Callback bei jedem Retry: (attempt, exception, delay)."""

    on_failure: Callable[[BaseException, int], None] | None = None
    """Callback bei endgültigem Fehlschlag: (exception, total_attempts)."""

    def calculate_delay(self, attempt: int) -> float:
        """
        Berechnet die Verzögerung für einen bestimmten Versuch.

        Args:
            attempt: Aktuelle Versuchsnummer (1-basiert).

        Returns:
            Verzögerung in Sekunden.
        """
        # Exponentielles Backoff
        delay = self.base_delay * (self.exponential_base ** (attempt - 1))

        # Maximum begrenzen
        delay = min(delay, self.max_delay)

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

        return max(0, delay)

    def should_retry(self, exception: BaseException, attempt: int) -> bool:
        """
        Prüft, ob ein weiterer Versuch unternommen werden sollte.

        Args:
            exception: Die aufgetretene Ausnahme.
            attempt: Aktuelle Versuchsnummer.

        Returns:
            True wenn Retry erfolgen soll.
        """
        # Maximale Versuche erreicht?
        if attempt >= self.max_attempts:
            return False

        # Ausnahme-Typ prüfen
        if not isinstance(exception, self.retry_on):
            return False

        # Custom-Prüfung
        if self.retry_if and not self.retry_if(exception):
            return False

        return True


@dataclass
class RetryStats:
    """
    Statistiken für Retry-Operationen.

    Sammelt Informationen über Erfolge, Fehlschläge und Retries.
    """

    function_name: str
    total_calls: int = 0
    successful_calls: int = 0
    failed_calls: int = 0
    total_retries: int = 0
    total_retry_time: float = 0.0
    last_failure: datetime | None = None
    last_failure_reason: str | None = None

    @property
    def success_rate(self) -> float:
        """Erfolgsrate in Prozent."""
        if self.total_calls == 0:
            return 100.0
        return (self.successful_calls / self.total_calls) * 100

    @property
    def avg_retries_per_call(self) -> float:
        """Durchschnittliche Retries pro Aufruf."""
        if self.total_calls == 0:
            return 0.0
        return self.total_retries / self.total_calls

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert die Statistiken in ein Dictionary."""
        return {
            "function": self.function_name,
            "total_calls": self.total_calls,
            "successful_calls": self.successful_calls,
            "failed_calls": self.failed_calls,
            "success_rate": self.success_rate,
            "total_retries": self.total_retries,
            "avg_retries_per_call": self.avg_retries_per_call,
            "total_retry_time_seconds": self.total_retry_time,
            "last_failure": (
                self.last_failure.isoformat() if self.last_failure else None
            ),
            "last_failure_reason": self.last_failure_reason,
        }


# Globaler Speicher für Retry-Statistiken
_retry_stats: dict[str, RetryStats] = {}


def get_retry_stats(function_name: str | None = None) -> dict[str, RetryStats]:
    """
    Gibt Retry-Statistiken zurück.

    Args:
        function_name: Optional spezifischer Funktionsname.

    Returns:
        Dictionary mit Statistiken.
    """
    if function_name:
        if function_name in _retry_stats:
            return {function_name: _retry_stats[function_name]}
        return {}
    return dict(_retry_stats)


def retry(
    max_attempts: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    exponential_base: float = 2.0,
    jitter: float = 0.1,
    retry_on: tuple[Type[BaseException], ...] | Type[BaseException] = (Exception,),
    retry_if: Callable[[BaseException], bool] | None = None,
    on_retry: Callable[[int, BaseException, float], None] | None = None,
    on_failure: Callable[[BaseException, int], None] | None = None,
    config: RetryConfig | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """
    Decorator für automatische Retry-Logik mit exponential Backoff.

    Args:
        max_attempts: Maximale Anzahl von Versuchen.
        base_delay: Basis-Verzögerung in Sekunden.
        max_delay: Maximale Verzögerung in Sekunden.
        exponential_base: Basis für exponentielles Backoff.
        jitter: Jitter-Faktor für Zufälligkeit.
        retry_on: Ausnahme(n), bei denen Retry erfolgen soll.
        retry_if: Optionale Funktion für Retry-Entscheidung.
        on_retry: Callback bei jedem Retry.
        on_failure: Callback bei endgültigem Fehlschlag.
        config: Optional vorkonfiguriertes RetryConfig.

    Returns:
        Decorator-Funktion.

    Example:
        @retry(max_attempts=3, retry_on=ConnectionError)
        def fetch_data():
            return requests.get(url).json()

        @retry(
            max_attempts=5,
            retry_if=lambda e: "temporary" in str(e).lower()
        )
        def unstable_operation():
            pass
    """
    # Config erstellen oder verwenden
    if config:
        cfg = config
    else:
        if isinstance(retry_on, type):
            retry_on = (retry_on,)
        cfg = RetryConfig(
            max_attempts=max_attempts,
            base_delay=base_delay,
            max_delay=max_delay,
            exponential_base=exponential_base,
            jitter=jitter,
            retry_on=retry_on,
            retry_if=retry_if,
            on_retry=on_retry,
            on_failure=on_failure,
        )

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        # Statistik-Objekt erstellen
        stats_name = func.__qualname__
        if stats_name not in _retry_stats:
            _retry_stats[stats_name] = RetryStats(function_name=stats_name)
        stats = _retry_stats[stats_name]

        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            stats.total_calls += 1
            last_exception: BaseException | None = None
            retry_count = 0
            total_delay = 0.0

            for attempt in range(1, cfg.max_attempts + 1):
                try:
                    result = func(*args, **kwargs)
                    stats.successful_calls += 1
                    stats.total_retries += retry_count
                    stats.total_retry_time += total_delay
                    return result

                except BaseException as e:
                    last_exception = e

                    if not cfg.should_retry(e, attempt):
                        break

                    # Verzögerung berechnen und warten
                    delay = cfg.calculate_delay(attempt)
                    total_delay += delay

                    # Callback aufrufen
                    if cfg.on_retry:
                        cfg.on_retry(attempt, e, delay)

                    retry_count += 1
                    time.sleep(delay)

            # Endgültiger Fehlschlag
            stats.failed_calls += 1
            stats.total_retries += retry_count
            stats.total_retry_time += total_delay
            stats.last_failure = datetime.now()
            stats.last_failure_reason = str(last_exception)

            if cfg.on_failure and last_exception:
                cfg.on_failure(last_exception, cfg.max_attempts)

            # Letzte Exception erneut werfen
            if last_exception:
                raise last_exception
            raise RuntimeError("Retry fehlgeschlagen ohne Exception")

        # Statistik-Zugriff über die Funktion
        wrapper.retry_stats = stats  # type: ignore

        return wrapper

    return decorator


def async_retry(
    max_attempts: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    exponential_base: float = 2.0,
    jitter: float = 0.1,
    retry_on: tuple[Type[BaseException], ...] | Type[BaseException] = (Exception,),
    retry_if: Callable[[BaseException], bool] | None = None,
    on_retry: Callable[[int, BaseException, float], None] | None = None,
    on_failure: Callable[[BaseException, int], None] | None = None,
    config: RetryConfig | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """
    Async-Decorator für Retry-Logik.

    Wie @retry, aber für async Funktionen.

    Args:
        max_attempts: Maximale Anzahl von Versuchen.
        base_delay: Basis-Verzögerung in Sekunden.
        max_delay: Maximale Verzögerung in Sekunden.
        exponential_base: Basis für exponentielles Backoff.
        jitter: Jitter-Faktor.
        retry_on: Ausnahmen für Retry.
        retry_if: Optionale Prüffunktion.
        on_retry: Callback bei Retry.
        on_failure: Callback bei Fehlschlag.
        config: Optionales RetryConfig.

    Returns:
        Decorator-Funktion.

    Example:
        @async_retry(max_attempts=3)
        async def fetch_async():
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as response:
                    return await response.json()
    """
    if config:
        cfg = config
    else:
        if isinstance(retry_on, type):
            retry_on = (retry_on,)
        cfg = RetryConfig(
            max_attempts=max_attempts,
            base_delay=base_delay,
            max_delay=max_delay,
            exponential_base=exponential_base,
            jitter=jitter,
            retry_on=retry_on,
            retry_if=retry_if,
            on_retry=on_retry,
            on_failure=on_failure,
        )

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        stats_name = func.__qualname__
        if stats_name not in _retry_stats:
            _retry_stats[stats_name] = RetryStats(function_name=stats_name)
        stats = _retry_stats[stats_name]

        @functools.wraps(func)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            stats.total_calls += 1
            last_exception: BaseException | None = None
            retry_count = 0
            total_delay = 0.0

            for attempt in range(1, cfg.max_attempts + 1):
                try:
                    result = await func(*args, **kwargs)
                    stats.successful_calls += 1
                    stats.total_retries += retry_count
                    stats.total_retry_time += total_delay
                    return result

                except BaseException as e:
                    last_exception = e

                    if not cfg.should_retry(e, attempt):
                        break

                    delay = cfg.calculate_delay(attempt)
                    total_delay += delay

                    if cfg.on_retry:
                        cfg.on_retry(attempt, e, delay)

                    retry_count += 1
                    await asyncio.sleep(delay)

            stats.failed_calls += 1
            stats.total_retries += retry_count
            stats.total_retry_time += total_delay
            stats.last_failure = datetime.now()
            stats.last_failure_reason = str(last_exception)

            if cfg.on_failure and last_exception:
                cfg.on_failure(last_exception, cfg.max_attempts)

            if last_exception:
                raise last_exception
            raise RuntimeError("Retry fehlgeschlagen ohne Exception")

        wrapper.retry_stats = stats  # type: ignore

        return wrapper

    return decorator


# Vordefinierte Retry-Konfigurationen
NETWORK_RETRY = RetryConfig(
    max_attempts=3,
    base_delay=1.0,
    max_delay=30.0,
    exponential_base=2.0,
    jitter=0.2,
    retry_on=(ConnectionError, TimeoutError, OSError),
)

DATABASE_RETRY = RetryConfig(
    max_attempts=5,
    base_delay=0.5,
    max_delay=10.0,
    exponential_base=1.5,
    jitter=0.1,
)

AGGRESSIVE_RETRY = RetryConfig(
    max_attempts=10,
    base_delay=0.1,
    max_delay=5.0,
    exponential_base=1.5,
    jitter=0.3,
)