如何在Python中构建支持自动重试机制的API客户端类-Python教程-PHP中文网

requests默认不重试且无超时，易因网络抖动、429限流、503服务不可用或超时失败；需用urllib3.Retry配置Session，指定status_forcelist、backoff_factor和allowed_methods实现可配置、分类型、带指数退避的可控重试。

如何在python中构建支持自动重试机制的api客户端类

为什么直接用 `requests` 发请求容易失败且难恢复

网络抖动、服务端限流、临时 503 或超时，这些在调用外部 API 时太常见了。如果每次失败都抛出 requests.exceptions.RequestException 就不管了，客户端会频繁中断；而手动在每个 get() / post() 外套 try/except + time.sleep()，又重复、难维护、不统一。

关键不是“要不要重试”，而是“重试是否可配置、是否区分错误类型、是否带退避”。比如连接超时值得重试，但 401 Unauthorized 绝对不该重试——否则可能反复暴露无效 token。

默认不重试：所有异常都立即冒泡
只对特定状态码重试：502、503、504、429
只对特定异常重试：ConnectTimeout、ReadTimeout、ConnectionError
避免重试：400、401、403、404 等客户端错误

用 `urllib3.Retry` 配合 `requests.Session` 实现底层重试策略

requests 本身不提供开箱即用的重试逻辑，但它底层依赖 urllib3，而 urllib3.Retry 是真正可控的重试引擎。直接改 Session.mount() 是最干净的做法，避免侵入业务代码。

注意：不能只靠 max_retries 参数，必须显式传入 status_forcelist 和 backoff_factor，否则 5xx 默认不重试，且无退避。

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
<p>session = requests.Session()
retry_strategy = Retry(
total=3,
status_forcelist=(502, 503, 504, 429),
backoff_factor=1,  # 第一次重试延迟 1s，第二次 2s，第三次 4s（指数退避）
allowed_methods=["HEAD", "GET", "OPTIONS", "POST"]  # 显式声明，避免 POST 被默认禁掉
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
</p>

⚠️ 容易踩的坑：allowed_methods 默认不含 POST，如果你调的是 RESTful 写接口，务必加上；backoff_factor 是乘数，不是固定秒数；total 包含首次请求，设为 3 表示最多发 4 次（1 次原请求 + 3 次重试）。

封装成客户端类时，如何把重试逻辑和业务逻辑解耦

不要在每个方法里写 for i in range(3): try ... except。重试是传输层关注的事，API 客户端类应该只暴露语义化方法（如 get_user(user_id)），内部复用同一个带重试的 session。

python-parallelization

将顺序 Python 代码转换为并行/并发实现，用于在需要并行化 Python 代码、提升性能时使用。

下载

建议把重试策略作为构造参数注入，方便测试（比如单元测试时传入 Retry(total=0) 关闭重试），也便于不同环境差异化配置（开发关重试，生产开）。

class APIClient:
    def __init__(self, base_url: str, retry_strategy: Retry = None):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        if retry_strategy is None:
            retry_strategy = Retry(total=2, status_forcelist=(502, 503, 504))
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)
<pre class="brush:php;toolbar:false;">def get_user(self, user_id: int) -> dict:
    resp = self.session.get(f"{self.base_url}/users/{user_id}")
    resp.raise_for_status()  # 这里才检查 4xx/5xx，确保重试已生效
    return resp.json()

重点：把 resp.raise_for_status() 放在重试之后，而不是之前；否则重试还没触发就提前抛错了。同时，不要在 get_user 里捕获异常再重试——那会覆盖底层重试机制。

重试时如何传递上下文（如 trace_id、重试次数）并避免敏感信息泄露

日志里看不到哪次请求是第几次重试，排查问题就很吃力；更麻烦的是，如果请求体含 token 或用户数据，重试时原样重发没问题，但若 body 是动态生成（如含时间戳或 nonce），重复提交可能导致幂等性问题。

用 session.hooks["response"] 注入日志钩子，记录重试次数：resp.request.headers.get("X-Retry-Count", "0")
对非幂等请求（如 POST /orders），优先考虑服务端支持 Idempotency-Key header，而不是依赖客户端重试
避免在重试时修改请求体——除非你明确控制了幂等逻辑；否则保持 body 不变最安全
敏感字段（如 Authorization）默认随重试自动带上，无需额外处理

真正棘手的点往往不在“怎么加重试”，而在“重试是否改变了语义”——比如一个扣款接口，重试三次，后端没做幂等，就真扣了三次。这种时候，重试机制反而成了 bug 放大器。