spb/api-ka

1# ============================================2# Projet   : API-KA3# Fichier  : src/utils/retry.py4# Node     : m3u96b5# Author   : Simon-Pierre Boucher6# Contact  : contact@spboucher.ai7# Date     : 2026-08-168# ============================================9"""Relance avec backoff exponentiel : 3 tentatives, délais 30s → 2min → 10min."""1011from __future__ import annotations1213import functools14import time15from collections.abc import Callable, Sequence16from typing import Any, TypeVar1718from src.utils.logger import get_logger1920T = TypeVar("T")2122DEFAULT_ATTEMPTS = 323DEFAULT_DELAYS: tuple[float, ...] = (30.0, 120.0, 600.0)242526def retry_call(27    func: Callable[[], T],28    *,29    attempts: int = DEFAULT_ATTEMPTS,30    delays: Sequence[float] = DEFAULT_DELAYS,31    on_retry: Callable[[int, BaseException, float], None] | None = None,32) -> T:33    """Exécute ``func`` avec jusqu'à ``attempts`` relances après l'essai initial.3435    Args:36        func: Callable sans argument à exécuter.37        attempts: Nombre maximal de relances après le premier échec.38        delays: Délais (secondes) avant chaque relance ; le dernier est réutilisé39            si ``attempts`` dépasse la longueur de la séquence.40        on_retry: Callback ``(numéro_de_relance, exception, délai)`` appelé41            avant chaque relance.4243    Returns:44        Le résultat de ``func``.4546    Raises:47        BaseException: La dernière exception si toutes les tentatives échouent.48    """49    logger = get_logger()50    for attempt in range(attempts + 1):51        try:52            return func()53        except Exception as exc:54            if attempt >= attempts:55                raise56            delay = float(delays[min(attempt, len(delays) - 1)]) if delays else 0.057            logger.warning(58                "Échec, relance planifiée",59                extra={60                    "attempt": attempt + 1,61                    "delay_seconds": delay,62                    "error": str(exc),63                },64            )65            if on_retry is not None:66                on_retry(attempt + 1, exc, delay)67            if delay > 0:68                time.sleep(delay)69    raise RuntimeError("retry_call: état inatteignable")  # pragma: no cover707172def retry(73    attempts: int = DEFAULT_ATTEMPTS,74    delays: Sequence[float] = DEFAULT_DELAYS,75) -> Callable[[Callable[..., T]], Callable[..., T]]:76    """Décorateur appliquant :func:`retry_call` à la fonction décorée."""7778    def decorator(func: Callable[..., T]) -> Callable[..., T]:79        @functools.wraps(func)80        def wrapper(*args: Any, **kwargs: Any) -> T:81            return retry_call(82                lambda: func(*args, **kwargs), attempts=attempts, delays=delays83            )8485        return wrapper8687    return decorator88