Modèles de Résilience Python

Construisez des applications Python tolérantes aux pannes qui gèrent gracieusement les défaillances transitoires, les problèmes réseau et les pannes de service. Les modèles de résilience gardent les systèmes opérationnels quand les dépendances sont peu fiables.

Quand utiliser cette compétence

Ajouter une logique de retry aux appels de services externes
Implémenter des timeouts pour les opérations réseau
Construire des microservices tolérants aux pannes
Gérer le rate limiting et la backpressure
Créer des décorateurs d'infrastructure
Concevoir des circuit breakers

Concepts Fondamentaux

1. Défaillances Transitoires vs Permanentes

Réessayez les erreurs transitoires (timeouts réseau, problèmes de service temporaires). N'essayez pas les erreurs permanentes (identifiants invalides, mauvaises requêtes).

2. Backoff Exponentiel

Augmentez le délai d'attente entre les tentatives pour éviter de surcharger les services en cours de récupération.

3. Jitter

Ajoutez de l'aléatoire au backoff pour éviter le thundering herd quand de nombreux clients font des retries simultanément.

4. Retries Bornés

Limitez à la fois le nombre de tentatives et la durée totale pour éviter les boucles de retry infinies.

Démarrage Rapide

from tenacity import retry, stop_after_attempt, wait_exponential_jitter

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential_jitter(initial=1, max=10),
)
def call_external_service(request: dict) -> dict:
    return httpx.post("https://api.example.com", json=request).json()

Modèles Fondamentaux

Modèle 1 : Retry Basique avec Tenacity

Utilisez la bibliothèque tenacity pour une logique de retry de qualité production. Pour les cas plus simples, envisagez une fonctionnalité de retry intégrée ou une implémentation personnalisée légère.

from tenacity import (
    retry,
    stop_after_attempt,
    stop_after_delay,
    wait_exponential_jitter,
    retry_if_exception_type,
)

TRANSIENT_ERRORS = (ConnectionError, TimeoutError, OSError)

@retry(
    retry=retry_if_exception_type(TRANSIENT_ERRORS),
    stop=stop_after_attempt(5) | stop_after_delay(60),
    wait=wait_exponential_jitter(initial=1, max=30),
)
def fetch_data(url: str) -> dict:
    """Récupérer les données avec retry automatique sur les défaillances transitoires."""
    response = httpx.get(url, timeout=30)
    response.raise_for_status()
    return response.json()

Modèle 2 : Retry Uniquement les Erreurs Appropriées

Whitelistez les exceptions transitoires spécifiques. Ne réessayez jamais :

ValueError, TypeError - Ce sont des bugs, pas des problèmes transitoires
AuthenticationError - Les identifiants invalides ne deviendront pas valides
Erreurs HTTP 4xx (sauf 429) - Les erreurs client sont permanentes

from tenacity import retry, retry_if_exception_type
import httpx

# Définir ce qui est retryable
RETRYABLE_EXCEPTIONS = (
    ConnectionError,
    TimeoutError,
    httpx.ConnectTimeout,
    httpx.ReadTimeout,
)

@retry(
    retry=retry_if_exception_type(RETRYABLE_EXCEPTIONS),
    stop=stop_after_attempt(3),
    wait=wait_exponential_jitter(initial=1, max=10),
)
def resilient_api_call(endpoint: str) -> dict:
    """Faire un appel API avec retry sur les problèmes réseau."""
    return httpx.get(endpoint, timeout=10).json()

Modèle 3 : Retries de Code de Statut HTTP

Réessayez les codes de statut HTTP spécifiques qui indiquent des problèmes transitoires.

from tenacity import retry, retry_if_result, stop_after_attempt
import httpx

RETRY_STATUS_CODES = {429, 502, 503, 504}

def should_retry_response(response: httpx.Response) -> bool:
    """Vérifier si la réponse indique une erreur retryable."""
    return response.status_code in RETRY_STATUS_CODES

@retry(
    retry=retry_if_result(should_retry_response),
    stop=stop_after_attempt(3),
    wait=wait_exponential_jitter(initial=1, max=10),
)
def http_request(method: str, url: str, **kwargs) -> httpx.Response:
    """Faire une requête HTTP avec retry sur les codes de statut transitoires."""
    return httpx.request(method, url, timeout=30, **kwargs)

Modèle 4 : Retry Combiné Exception et Statut

Gérez à la fois les exceptions réseau et les codes de statut HTTP.

from tenacity import (
    retry,
    retry_if_exception_type,
    retry_if_result,
    stop_after_attempt,
    wait_exponential_jitter,
    before_sleep_log,
)
import logging
import httpx

logger = logging.getLogger(__name__)

TRANSIENT_EXCEPTIONS = (
    ConnectionError,
    TimeoutError,
    httpx.ConnectError,
    httpx.ReadTimeout,
)
RETRY_STATUS_CODES = {429, 500, 502, 503, 504}

def is_retryable_response(response: httpx.Response) -> bool:
    return response.status_code in RETRY_STATUS_CODES

@retry(
    retry=(
        retry_if_exception_type(TRANSIENT_EXCEPTIONS) |
        retry_if_result(is_retryable_response)
    ),
    stop=stop_after_attempt(5),
    wait=wait_exponential_jitter(initial=1, max=30),
    before_sleep=before_sleep_log(logger, logging.WARNING),
)
def robust_http_call(
    method: str,
    url: str,
    **kwargs,
) -> httpx.Response:
    """Appel HTTP avec gestion de retry complète."""
    return httpx.request(method, url, timeout=30, **kwargs)

Modèles Avancés

Modèle 5 : Logging des Tentatives de Retry

Suivez le comportement de retry pour le débogage et l'alerting.

from tenacity import retry, stop_after_attempt, wait_exponential
import structlog

logger = structlog.get_logger()

def log_retry_attempt(retry_state):
    """Logger les informations de retry détaillées."""
    exception = retry_state.outcome.exception()
    logger.warning(
        "Retrying operation",
        attempt=retry_state.attempt_number,
        exception_type=type(exception).__name__,
        exception_message=str(exception),
        next_wait_seconds=retry_state.next_action.sleep if retry_state.next_action else None,
    )

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, max=10),
    before_sleep=log_retry_attempt,
)
def call_with_logging(request: dict) -> dict:
    """Appel externe avec logging de retry."""
    ...

Modèle 6 : Décorateur Timeout

Créez des décorateurs timeout réutilisables pour une gestion cohérente des timeouts.

import asyncio
from functools import wraps
from typing import TypeVar, Callable

T = TypeVar("T")

def with_timeout(seconds: float):
    """Décorateur pour ajouter un timeout aux fonctions async."""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        async def wrapper(*args, **kwargs) -> T:
            return await asyncio.wait_for(
                func(*args, **kwargs),
                timeout=seconds,
            )
        return wrapper
    return decorator

@with_timeout(30)
async def fetch_with_timeout(url: str) -> dict:
    """Récupérer une URL avec timeout de 30 secondes."""
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

Modèle 7 : Préoccupations Transversales via Décorateurs

Empilez les décorateurs pour séparer l'infrastructure de la logique métier.

from functools import wraps
from typing import TypeVar, Callable
import structlog

logger = structlog.get_logger()
T = TypeVar("T")

def traced(name: str | None = None):
    """Ajouter du tracing aux appels de fonction."""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        span_name = name or func.__name__

        @wraps(func)
        async def wrapper(*args, **kwargs) -> T:
            logger.info("Operation started", operation=span_name)
            try:
                result = await func(*args, **kwargs)
                logger.info("Operation completed", operation=span_name)
                return result
            except Exception as e:
                logger.error("Operation failed", operation=span_name, error=str(e))
                raise
        return wrapper
    return decorator

# Empiler plusieurs préoccupations
@traced("fetch_user_data")
@with_timeout(30)
@retry(stop=stop_after_attempt(3), wait=wait_exponential_jitter())
async def fetch_user_data(user_id: str) -> dict:
    """Récupérer l'utilisateur avec tracing, timeout et retry."""
    ...

Modèle 8 : Dependency Injection pour la Testabilité

Passez les composants d'infrastructure via les constructeurs pour un test facile.

from dataclasses import dataclass
from typing import Protocol

class Logger(Protocol):
    def info(self, msg: str, **kwargs) -> None: ...
    def error(self, msg: str, **kwargs) -> None: ...

class MetricsClient(Protocol):
    def increment(self, metric: str, tags: dict | None = None) -> None: ...
    def timing(self, metric: str, value: float) -> None: ...

@dataclass
class UserService:
    """Service avec infrastructure injectée."""

    repository: UserRepository
    logger: Logger
    metrics: MetricsClient

    async def get_user(self, user_id: str) -> User:
        self.logger.info("Fetching user", user_id=user_id)
        start = time.perf_counter()

        try:
            user = await self.repository.get(user_id)
            self.metrics.increment("user.fetch.success")
            return user
        except Exception as e:
            self.metrics.increment("user.fetch.error")
            self.logger.error("Failed to fetch user", user_id=user_id, error=str(e))
            raise
        finally:
            elapsed = time.perf_counter() - start
            self.metrics.timing("user.fetch.duration", elapsed)

# Facile à tester avec des fakes
service = UserService(
    repository=FakeRepository(),
    logger=FakeLogger(),
    metrics=FakeMetrics(),
)

Modèle 9 : Defaults Fail-Safe

Dégradez gracieusement quand les opérations non-critiques échouent.

from typing import TypeVar
from collections.abc import Callable

T = TypeVar("T")

def fail_safe(default: T, log_failure: bool = True):
    """Retourner la valeur par défaut en cas d'échec au lieu de lever."""
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        async def wrapper(*args, **kwargs) -> T:
            try:
                return await func(*args, **kwargs)
            except Exception as e:
                if log_failure:
                    logger.warning(
                        "Operation failed, using default",
                        function=func.__name__,
                        error=str(e),
                    )
                return default
        return wrapper
    return decorator

@fail_safe(default=[])
async def get_recommendations(user_id: str) -> list[str]:
    """Obtenir les recommandations, retourner une liste vide en cas d'échec."""
    ...

Résumé des Bonnes Pratiques

Retry uniquement les erreurs transitoires - Ne réessayez pas les bugs ou les défaillances d'authentification
Utilisez le backoff exponentiel - Donnez aux services le temps de récupérer
Ajoutez du jitter - Évitez le thundering herd des retries synchronisés
Limitez la durée totale - stop_after_attempt(5) | stop_after_delay(60)
Logger chaque retry - Les retries silencieux cachent les problèmes systémiques
Utilisez les décorateurs - Séparez la logique de retry de la logique métier
Injectez les dépendances - Rendez l'infrastructure testable
Définissez des timeouts partout - Chaque appel réseau a besoin d'un timeout
Échouez gracieusement - Retournez des valeurs en cache/par défaut pour les chemins non-critiques
Surveillez les taux de retry - Des taux de retry élevés indiquent des problèmes sous-jacents

python-resilience