Python requests使用urllib3.Retry重试抛异常后如何获取对应HTTP状态码
解决方案
下面提供两种无需引入额外第三方库的实现方式:
方案1:直接从异常链提取(无需修改原有重试逻辑)
该方案适合不想调整现有重试封装逻辑的场景,通过逐层解析异常对象属性获取状态码,额外提供正则兜底兼容不同版本的urllib3结构差异:
import requests from requests.exceptions import RetryError import re # 调用你原有封装的重试session session = session_with_retries() try: resp = session.get("你的请求地址") except RetryError as e: final_status = None # 第一层:从requests的RetryError中取出urllib3的MaxRetryError实例 max_retry_err = e.args[0] # 第二层:取出异常原因对象 reason = max_retry_err.reason # 优先从属性中直接取状态码 if hasattr(reason, 'status'): final_status = reason.status # 兜底方案:从异常信息中正则匹配状态码 else: err_msg = str(reason) match_res = re.search(r'too many (\d+) error responses', err_msg) if match_res: final_status = int(match_res.group(1)) if final_status: print(f"最终触发重试超限的状态码为:{final_status}")
方案2:自定义Retry子类(稳定性更高,兼容性更强)
该方案通过继承urllib3原生的Retry类,新增属性记录最后一次请求的状态码,不受urllib3版本迭代导致的异常结构变化影响:
步骤1:自定义重试类,新增状态码记录逻辑
from urllib3 import Retry class RetryWithLastStatus(Retry): # 新增属性存储最后一次请求的HTTP状态码 last_status = None def increment(self, method=None, url=None, response=None, error=None, _pool=None, _stacktrace=None): # 每次触发重试前,记录当前请求的状态码 if response is not None: self.last_status = response.status # 执行原生重试逻辑 return super().increment(method, url, response, error, _pool, _stacktrace)
步骤2:替换原有重试逻辑中的Retry类
修改你原有session_with_retries函数中重试实例的初始化代码,将原生Retry替换为自定义的RetryWithLastStatus即可,其余配置无需改动:
def session_with_retries(max_retries: int = DEFAULT_RETRIES, backoff_factor: float = DEFAULT_BACKOFF, proxies: dict = None) -> requests.Session: new_session = requests.Session() # 替换为自定义的重试类 retries = RetryWithLastStatus( total=max_retries, connect=max_retries, read=max_retries, status=max_retries, allowed_methods=frozenset(['HEAD', 'GET', 'POST']), status_forcelist=frozenset([500, 502, 503, 504]), backoff_factor=backoff_factor, ) # 后续逻辑保持不变 retry_adapter = HTTPAdapter(max_retries=retries) new_session.mount('http://', retry_adapter) new_session.mount('https://', retry_adapter) if proxies is not None: new_session.proxies.update(proxies) return new_session
步骤3:异常捕获时直接取存储的状态码
try: resp = session.get("你的请求地址") except RetryError as e: max_retry_err = e.args[0] final_status = max_retry_err.reason.last_retry.last_status print(f"最终触发重试超限的状态码为:{final_status}")
内容的提问来源于stack exchange,提问作者John Byro
相关产品推荐
相关产品推荐

