如何为Python Requests创建自动重试的请求中间件?
可行,以下是具体实现方案
你的需求完全可以实现,且不需要改动request1.py和request2.py里的原有请求代码,下面提供两种实用方案:
方案1:用 Requests 原生 HTTPAdapter 实现全局重试(推荐)
Requests 内置了基于urllib3的重试支持,通过配置HTTPAdapter并全局挂载,就能让所有请求自动触发重试逻辑。
middleware.py 代码实现:
import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def setup_global_retry(): # 定义重试规则 retry_strategy = Retry( total=3, # 最多重试3次 backoff_factor=1, # 重试间隔递增:第1次等1s,第2次2s,第3次4s status_forcelist=[429, 500, 502, 503, 504], # 遇到这些HTTP状态码时重试 allowed_methods=["GET", "POST"] # 允许重试的请求方法 ) # 创建适配器并挂载到HTTP/HTTPS请求 adapter = HTTPAdapter(max_retries=retry_strategy) session = requests.session() session.mount("http://", adapter) session.mount("https://", adapter) # 替换全局requests的默认会话 requests.Session = lambda: session # 项目启动时执行初始化 setup_global_retry()
使用方式:在项目的入口文件(比如main.py)里先导入middleware,之后所有requests.get/post的请求都会自动应用重试策略,request1、request2变量会自动拿到成功后的响应。
方案2:猴子补丁替换请求方法(自定义异常捕获)
如果需要更灵活地控制捕获的异常类型(比如只针对连接超时、连接错误重试),可以用猴子补丁直接替换requests的请求方法,包裹重试逻辑。
middleware.py 代码实现:
import requests from requests.exceptions import ConnectionError, Timeout, HTTPError # 保存原始请求方法 _original_get = requests.get _original_post = requests.post def _retry_wrapper(func, *args, **kwargs): max_retries = 3 current_retry = 0 while current_retry < max_retries: try: resp = func(*args, **kwargs) resp.raise_for_status() # 主动抛出HTTP错误(比如4xx/5xx) return resp except (ConnectionError, Timeout, HTTPError) as e: current_retry += 1 if current_retry == max_retries: # 达到重试上限,抛出原异常 raise e # 替换原始方法 def patched_get(*args, **kwargs): return _retry_wrapper(_original_get, *args, **kwargs) def patched_post(*args, **kwargs): return _retry_wrapper(_original_post, *args, **kwargs) requests.get = patched_get requests.post = patched_post
使用方式:同样在项目入口导入middleware,原有请求代码无需任何修改即可自动重试。
关键注意事项
- 禁止无限重试:必须设置最大重试次数,避免在API彻底不可用时陷入死循环。
- 幂等性检查:POST请求要确保接口是幂等的(重复提交不会生成重复数据),否则重试可能导致业务异常。
- 退避策略:加入递增间隔(比如
backoff_factor),避免短时间内频繁请求触发API限流。
内容的提问来源于stack exchange,提问作者Lirit
相关产品推荐
相关产品推荐

