Padrões de Resiliência em Python

Crie aplicações Python tolerantes a falhas que lidam graciosamente com falhas transitórias, problemas de rede e interrupções de serviço. Padrões de resiliência mantêm os sistemas funcionando quando as dependências são não confiáveis.

Quando Usar Esta Habilidade

• Adicionar lógica de retentativa a chamadas de serviço externas
• Implementar timeouts para operações de rede
• Construir microsserviços tolerantes a falhas
• Lidar com limitação de taxa e backpressure
• Criar decoradores de infraestrutura
• Projetar circuit breakers

Conceitos Principais

1. Falhas Transitórias vs. Permanentes

Retente erros transitórios (timeouts de rede, problemas temporários de serviço). Não retente erros permanentes (credenciais inválidas, requisições ruins).

2. Backoff Exponencial

Aumente o tempo de espera entre as retentativas para evitar sobrecarregar serviços em recuperação.

3. Jitter

Adicione aleatoriedade ao backoff para evitar o "thundering herd" (manada trovejante) quando muitos clientes tentam novamente simultaneamente.

4. Retentativas Limitadas

Limite tanto o número de tentativas quanto a duração total para evitar loops infinitos de retentativa.

Início Rápido

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()

Padrões Fundamentais

Padrão 1: Retentativa Básica com Tenacity

Use a biblioteca tenacity para lógica de retentativa de nível de produção. Para casos mais simples, considere a funcionalidade de retentativa integrada ou uma implementação customizada leve.

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:
    """Fetch data with automatic retry on transient failures."""
    response = httpx.get(url, timeout=30)
    response.raise_for_status()
    return response.json()

Padrão 2: Retentar Apenas Erros Apropriados

Liste especificamente exceções transitórias. Nunca retente:

• ValueError, TypeError - Estes são bugs, não problemas transitórios
• AuthenticationError - Credenciais inválidas não se tornarão válidas
• Erros HTTP 4xx (exceto 429) - Erros do cliente são permanentes

from tenacity import retry, retry_if_exception_type
import httpx

# Define what's 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:
    """Make API call with retry on network issues."""
    return httpx.get(endpoint, timeout=10).json()

Padrão 3: Retentativas de Código de Status HTTP

Retente códigos de status HTTP específicos que indicam problemas transitórios.

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:
    """Check if response indicates a retryable error."""
    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:
    """Make HTTP request with retry on transient status codes."""
    return httpx.request(method, url, timeout=30, **kwargs)

Padrão 4: Retentativa Combinada de Exceção e Status

Lide com exceções de rede e códigos de status 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)
    ),

…