tenacity
tenacity adiciona lógica de retentativa configurável a qualquer chamável - backoff exponencial, jitter, condições de parada e hooks - sem precisar criar loops while manualmente em chamadas de rede ou banco de dados instáveis.
Busque em todas as páginas da documentação
tenacity adiciona lógica de retentativa configurável a qualquer chamável - backoff exponencial, jitter, condições de parada e hooks - sem precisar criar loops while manualmente em chamadas de rede ou banco de dados instáveis.
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 fetch_status() -> dict:
response = httpx.get("https://api.example.com/health", timeout=5.0)
response.raise_for_status()
return response.json()Quando usar isso:
from __future__ import annotations
import logging
import httpx
from tenacity import (
before_sleep_log,
retry,
retry_if_exception_type,
stop_after_attempt,
wait_exponential_jitter,
)
logger = logging.getLogger(__name__)
class TransientError(Exception):
pass
def _raise_for_transient(response: httpx.Response) -> None:
if response.status_code >= 500:
raise TransientError(f"server error {response.status_code}")
response.raise_for_status()
@retry(
retry=retry_if_exception_type((TransientError, httpx.TimeoutException)),
stop=stop_after_attempt(5),
wait=wait_exponential_jitter(initial=0.5, max=8),
before_sleep=before_sleep_log(logger, logging.WARNING),
reraise=True,
)
def charge_customer(customer_id: str, amount_cents: int) -> str:
with httpx.Client(timeout=10.0) as client:
response = client.post(
"/charges",
json={"customer_id": customer_id, "amount_cents": amount_cents},
headers={"Idempotency-Key": f"{customer_id}:{amount_cents}"},
)
_raise_for_transient(response)
return response.json()["id"]
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
print(charge_customer("cus_1", 1999))O que isso demonstra:
TransientError e timeouts - não para erros 4xx do clientebefore_sleep_log exibe as tentativas de retentativa nos logsreraise=True preserva a exceção final após o esgotamento@retry envolve funções síncronas; use @retry em funções assíncronas com espera/parada compatíveis com async (ou o contexto AsyncRetrying).stop_after_attempt, stop_after_delay, stop_any.retry_if_exception, retry_if_result, retry_unless_exception_type.| Tipo de Erro | Retentar? | Notas |
|---|---|---|
| HTTP 408/429/502/503/504 | Frequentemente | Respeitar Retry-After quando presente |
| HTTP 400/401/404 | Não | Corrigir requisição ou autenticação |
| Deadlock de DB | Sim | Backoff curto, tentativas limitadas |
| ValidationError | Não | Os dados não mudarão na retentativa |
from tenacity import Retrying
# Estilo imperativo dentro do corpo de um loop
for attempt in Retrying(stop=stop_after_attempt(3)):
with attempt:
do_work()stop_after_attempt mais alerta em caso de falha final.Exception genérico - Mascara bugs de programação. Correção: estreitar retry_if_exception_type.wait_exponential_jitter.reraise=True e métricas na contagem de retentativas.| Alternativa | Usar Quando | Não Usar Quando |
|---|---|---|
Loop for manual | Scripts pontuais com 2 retentativas | Políticas compartilhadas entre serviços |
urllib3.util.retry | Apenas na pilha requests/urllib3 | Aplicativos nativos httpx |
Celery autoretry_for | Retentativas de fila de tarefas no nível do worker | Chamadas de cliente HTTP no processo |
| Circuit breaker (pybreaker) | Falha sustentada do upstream | Apenas falhas transitórias breves |
Sim - use helpers de retentativa async ou AsyncRetrying dependendo da versão do tenacity; mantenha um loop de eventos por cliente.
Lance um TransientError de domínio após inspecionar response.status_code dentro da função envolvida.
Use retry_error_callback ou envolva a chamada e trate a exceção relançada no chamador.
Comece com 3-5 para HTTP; ajuste com base na latência p99 e SLOs do upstream - registre métricas de retentativa para decidir.
Ambas as camadas são válidas - retentativas do cliente lidam com falhas de subsegundo; retentativas do Celery lidam com reinícios de processo.
Simule a dependência instável para falhar duas vezes e depois ter sucesso; afirme que a contagem de chamadas é igual a três.
Não automaticamente - analise o cabeçalho em before_sleep e retorne uma espera personalizada, se necessário.
Passe uma política de retentativa no-op via injeção de dependência ou defina uma flag de ambiente que retorne a função nua.
Retente no limite da unidade de trabalho após session.rollback() - não retente no meio da transação.
O decorador é reentrante por chamada; compartilhe nenhum estado mutável dentro de funções retentadas sem locks.
Versões da Stack: Esta página foi escrita para Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, e uv 0.6+.
Revisado por Chris St. John·Última atualização: 19 de jul. de 2026