Исходный код tbank.core.retry
from __future__ import annotations
import random
from dataclasses import dataclass
from typing import FrozenSet, Optional
# Методы, которые безопасно повторять: запрос можно отправить дважды
# без побочного эффекта (в отличие от POST-платежа).
IDEMPOTENT_METHODS: FrozenSet[str] = frozenset(
{"GET", "HEAD", "OPTIONS", "PUT", "DELETE"}
)
[документация]
@dataclass(frozen=True)
class RetryPolicy:
attempts: int = 3
backoff_base: float = 0.5
backoff_max: float = 8.0
jitter: bool = True
retry_statuses: FrozenSet[int] = frozenset({429, 500, 502, 503, 504})
# True → ретраить и неидемпотентные запросы (POST без Idempotency-Key).
# По умолчанию выключено: таймаут POST-платежа не означает, что платёж
# не прошёл, и повтор может продублировать операцию.
retry_non_idempotent: bool = False
[документация]
def should_retry(
policy: RetryPolicy,
*,
status: Optional[int],
attempt: int,
method: str = "GET",
idempotent: Optional[bool] = None,
) -> bool:
"""Ретраить ли попытку номер `attempt` (1-based).
`idempotent` — явный признак безопасности повтора (например, у запроса
есть Idempotency-Key); если не задан, выводится из метода.
429 повторяется всегда: сервер отклонил запрос до обработки.
"""
if attempt >= policy.attempts:
return False
if idempotent is None:
idempotent = method.upper() in IDEMPOTENT_METHODS
safe = idempotent or policy.retry_non_idempotent
if status is None: # сетевая ошибка/таймаут
return safe
if status not in policy.retry_statuses:
return False
return safe or status == 429
[документация]
def compute_delay(
policy: RetryPolicy, *, attempt: int, retry_after: Optional[float] = None
) -> float:
"""Задержка перед попыткой `attempt` (1-based)."""
if retry_after is not None:
return min(retry_after, policy.backoff_max)
delay = min(policy.backoff_base * (2.0 ** (attempt - 1)), policy.backoff_max)
if policy.jitter:
delay *= 0.5 + random.random() / 2
return delay