通用重试装饰器默认重试所有异常,可能重复有副作用的操作
原文依据:3 处参考实现默认捕获 Exception,并在失败后再次调用任意传入函数;它没有区分临时故障、验证失败或操作是否幂等。
若代理把该模式用于扣款、写入、发送消息或创建资源,第一次操作实际成功但响应失败时,重试可能造成重复扣款、重复记录或重复通知。
装饰器的默认异常范围是 Exception,失败后会再次调用任意传入函数,确实可能重复付款、写入或发送消息等非幂等副作用。示例用法把范围收窄到 NetworkError,降低了该具体示例的风险,但默认接口仍支持过宽重试。风险只在用户采用默认值或用于非幂等操作时出现;可要求仅重试明确的瞬时错误,并加入幂等键或重试安全约束。
def retry( max_attempts: int = 3, backoff_factor: float = 2.0, exceptions: tuple = (Exception,)): """Retry decorator with exponential backoff.""" def decorator(func: Callable[..., T]) -> Callable[..., T]: @wraps(func) def wrapper(*args, **kwargs) -> T: last_exception = None for attempt in range(max_attempts): try: return func(*args, **kwargs) except exceptions as e: last_exception = e if attempt < max_attempts - 1: sleep_time = backoff_factor ** attempt time.sleep(sleep_time) continue raise raise last_exception查看另外 2 个位置
last_exception = None for attempt in range(max_attempts): try: return func(*args, **kwargs) except exceptions as e: last_exception = e if attempt < max_attempts - 1: sleep_time = backoff_factor ** attempt time.sleep(sleep_time) continue raise raise last_exception# Usage@retry(max_attempts=3, exceptions=(NetworkError,))def fetch_data(url: str) -> dict: response = requests.get(url, timeout=5) response.raise_for_status() return response.json()