如何用Python编写高可靠性的自动重试装饰器以处理不稳定的网络API？-Python教程-PHP中文网

直接用 time.sleep() 重试会失败，因其无法处理超时中断、未区分异常类型、缺乏指数退避且易无限等待；tenacity 是最稳妥方案，支持正交策略组合、异步重试及业务语义定制。

如何用python编写高可靠性的自动重试装饰器以处理不稳定的网络api？

为什么直接用 `time.sleep()` 做重试会失败

网络请求失败时，简单加个 time.sleep(1) 再循环，看似能“重试”，但实际会掩盖三类关键问题：超时未中断（卡住整个线程）、异常类型未区分（比如 401 和 503 完全不该同样重试）、指数退避缺失（连续猛冲可能触发限流）。更糟的是，没设置最大重试次数时，requests.exceptions.ConnectionError 可能导致无限等待。

`tenacity` 是目前最稳妥的重试方案

自己手写带退避、异常过滤、超时控制的重试逻辑极易出错。用 tenacity 库（非 retrying，后者已停更且不支持异步）是当前 Python 生态中最可靠的选择。它把重试策略拆解为正交维度，可组合、可测试、可监控。

安装与基础用法：

pip install tenacity

典型配置示例（处理常见网络异常）：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests
<p>@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=1, max=10),
retry=retry_if_exception_type((
requests.exceptions.Timeout,
requests.exceptions.ConnectionError,
requests.exceptions.HTTPError
))
)
def fetch_data(url):
resp = requests.get(url, timeout=5)
resp.raise_for_status()
return resp.json()</p>

stop_after_attempt(3)：最多执行 3 次（含首次），不是“重试 3 次”
wait_exponential 中 min=1 单位是秒，避免首重试过快；max=10 防止退避时间失控
只对指定异常重试，HTTPError 需显式加入——因为 raise_for_status() 抛出的就是它，而默认不捕获

必须手动处理的 HTTP 状态码边界情况

装饰器无法自动判断业务语义。比如 429 Too Many Requests 要看响应头 Retry-After，503 Service Unavailable 可能需退避，但 401 Unauthorized 或 404 Not Found 重试毫无意义。

macOS UI 自动化技能包 —— 用 Python 控制你的 Mac

让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限

下载

正确做法是：在函数内部捕获 HTTPError，解析状态码，再主动抛出定制异常供 tenacity 过滤：

from requests.exceptions import HTTPError
from tenacity import retry, retry_if_exception_type
<p>class TransientHTTPError(Exception):
"""仅用于标记可重试的 HTTP 错误"""
pass</p><p>@retry(retry=retry_if_exception_type(TransientHTTPError))
def fetch_with_backoff(url):
try:
resp = requests.get(url, timeout=5)
resp.raise_for_status()
return resp.json()
except HTTPError as e:
if e.response.status_code in (429, 503, 504):</p><h1>解析 Retry-After，或 fallback 到固定退避</h1><pre class="brush:python;toolbar:false;">        raise TransientHTTPError(str(e)) from e
    else:
        raise  # 其他状态码直接冒泡，不重试

异步请求下别误用同步装饰器

如果用 aiohttp 或 httpx.AsyncClient，直接套用上面的 @retry 会报 RuntimeWarning: coroutine 'xxx' was never awaited——因为 tenacity 默认同步执行，不会 await 协程。

必须换用 tenacity 的异步版本：

from tenacity import AsyncRetrying, stop_after_attempt, wait_exponential
import httpx
<p>async def fetch_async(url):
async for attempt in AsyncRetrying(
stop=stop_after_attempt(3),
wait=wait_exponential(min=1, max=10)
):
with attempt:
async with httpx.AsyncClient() as client:
resp = await client.get(url, timeout=5)
resp.raise_for_status()
return resp.json()</p>

不能用装饰器语法，必须用 async for attempt in AsyncRetrying(...) + with attempt:
AsyncRetrying 不支持 retry_if_exception_type 的异步变体，需在 with attempt: 块内手动 raise 触发重试
每次循环都新建 AsyncClient 实例是安全的；若复用 client，需确保其生命周期覆盖所有重试

真正的难点不在写几行装饰器，而在于精准识别哪些错误属于“瞬时失败”、哪些是“永久失败”，以及是否要根据响应内容（如 Retry-After、X-RateLimit-Remaining）动态调整策略——这些必须结合具体 API 文档做判断，没有通用解。