You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.24 06:15:05