怎样在Python中设计一个带重试机制的自动化API调用脚本？-Python教程-PHP中文网

直接用requests.get()不够用，因其无法应对网络抖动、服务端限流或临时超时等常见问题，需借助tenacity实现带退避策略、异常分类和结果校验的可控重试，或手动封装轻量版并注意超时、状态码判断、日志与耗时控制。

怎样在python中设计一个带重试机制的自动化api调用脚本？

为什么直接用 `requests.get()` 不够用

因为网络抖动、服务端限流、临时超时这些情况太常见了，一次失败就中断脚本，既不可靠也不符合生产场景。比如调用某个支付回调接口，偶尔返回 502 Bad Gateway 或 ConnectionError，这时候需要自动重试，但不能无脑循环——得控制次数、间隔和退出条件。

用 `tenacity` 实现可配置的重试逻辑

比手写 while + sleep 更可靠：它内置退避策略、异常分类、结果校验等能力，且支持同步/异步。安装后直接装饰函数即可：

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.ConnectionError, requests.exceptions.Timeout))
)
def call_api(url):
return requests.get(url, timeout=5)
</p>

注意几个关键点：

stop_after_attempt(3) 表示最多尝试 3 次（含首次），不是“重试 3 次”
wait_exponential 的 min=1 是首次失败后等待 1 秒，下次翻倍，上限卡在 10 秒，避免雪崩式重试
只对 ConnectionError 和 Timeout 重试，400 或 401 这类客户端错误不重试——它们不会因重试而变好
如果函数返回值需校验（比如必须有 "status": "success"），可用 retry_if_result 或自定义判断函数

手动实现轻量版重试（不用第三方库）

当项目不允许引入新依赖，或只需简单逻辑时，自己封装更可控：

import time
import requests
<p>def call_api_with_retry(url, max_retries=3, base_delay=1):
for i in range(max_retries + 1):
try:
resp = requests.get(url, timeout=5)
if resp.status_code == 200:
return resp.json()
elif resp.status_code in (429, 500, 502, 503, 504):
raise Exception(f"server error {resp.status_code}")
else:
return {"error": "client_error", "status_code": resp.status_code}
except (requests.exceptions.RequestException, ValueError):
if i == max_retries:
raise
time.sleep(base_delay * (2 ** i))  # 指数退避
return None
</p>

容易踩的坑：

361SCAN security check your skill python scripts

扫描并分析已安装技能，用于查看指定技能目录的名称、描述和详情，或列出所有已安装的技能。

下载

忘记把 timeout 传给 requests，导致某次请求卡死几十秒，拖垮整个重试节奏
对所有 HTTP 状态码都重试，比如 404 重试毫无意义
没限制总耗时，连续重试 3 次 + 指数退避可能耗时超过 1 分钟，而业务要求 10 秒内必须返回
没处理 JSON 解析失败（ValueError），导致重试被跳过

重试时如何传递上下文并记录日志

真实脚本里你得知道哪次失败、为什么失败、用了多久——否则出问题没法排查：

import logging
import time
<p>logger = logging.getLogger(<strong>name</strong>)</p><p>def call_api_with_logging(url, max_retries=3):
start_time = time.time()
for attempt in range(max_retries + 1):
try:
resp = requests.get(url, timeout=5)
elapsed = time.time() - start_time
logger.info(f"API success on attempt {attempt}, took {elapsed:.2f}s, status={resp.status_code}")
return resp
except Exception as e:
elapsed = time.time() - start_time
logger.warning(f"Attempt {attempt} failed: {e} (after {elapsed:.2f}s)")
if attempt == max_retries:
logger.error(f"Give up after {max_retries} retries")
raise
time.sleep(min(2 ** attempt, 10))
</p>

关键细节：