"""简单的重试工具:指数退避 + 抖动。 课程设计场景不引入额外依赖(如 tenacity),保持轻量。 """ from __future__ import annotations import random import time from typing import Callable, TypeVar, Tuple T = TypeVar("T") def retry( *, max_retries: int = 3, initial_delay: float = 1.0, backoff_factor: float = 2.0, max_delay: float = 10.0, jitter: float = 0.2, retry_exceptions: Tuple[type[BaseException], ...] = (Exception,), ): """重试装饰器。 - 第一次失败后等待 initial_delay - 之后按照 backoff_factor 指数增长 - 加一点 jitter 防止固定间隔 """ def decorator(func: Callable[..., T]) -> Callable[..., T]: def wrapper(*args, **kwargs) -> T: delay = initial_delay last_exc: BaseException | None = None for attempt in range(max_retries + 1): try: return func(*args, **kwargs) except retry_exceptions as exc: # noqa: PERF203 last_exc = exc if attempt >= max_retries: raise # jitter:delay*(1±jitter) factor = 1.0 + random.uniform(-jitter, jitter) sleep_s = min(max_delay, max(0.0, delay * factor)) time.sleep(sleep_s) delay = min(max_delay, delay * backoff_factor) # 理论上不会走到这里 assert last_exc is not None raise last_exc return wrapper return decorator