Requests错误与异常处理 | Python HTTP请求错误处理教程

为什么需要错误处理？

网络不可靠

网络连接可能随时中断，DNS解析可能失败，服务器可能宕机。

服务器错误

服务器可能返回错误状态码(4xx, 5xx)，或者响应格式不符合预期。

资源限制

请求可能因为超时、频率限制或认证失败而被拒绝。

错误处理的重要性

1

提高应用稳定性

良好的错误处理可以防止单个请求失败导致整个应用崩溃。

2

改善用户体验

向用户提供清晰的错误信息，而不是让应用无声地失败。

3

便于调试和维护

详细的错误日志有助于快速定位和修复问题。

4

实现优雅降级

当主要服务不可用时，可以提供备选方案或缓存数据。

Requests异常继承体系

Requests库中的所有异常都继承自requests.exceptions.RequestException，而它又继承自Python标准的IOError。

BaseException
│
└── Exception
    │
    └── IOError (Python标准异常)
        │
        └── RequestException (所有Requests异常的基类)
            ├── ConnectionError
            │    ├── ConnectTimeout
            │    ├── ProxyError
            │    └── SSLError
            ├── Timeout
            │    ├── ConnectTimeout
            │    └── ReadTimeout
            ├── URLRequired
            ├── TooManyRedirects
            ├── HTTPError
            └── ContentDecodingError

理解这个继承体系很重要，因为它允许你：

捕获特定类型的异常进行特殊处理
捕获基类异常进行通用处理
按照异常的重要性进行分级处理

常见异常类型

异常类	触发条件	说明	严重程度
`ConnectionError`	网络连接失败	DNS解析失败、拒绝连接、网络不可达等	严重
`HTTPError`	HTTP状态码4xx或5xx	服务器返回错误状态码，如404、500等	警告
`Timeout`	请求超时	连接超时或读取响应超时	警告
`TooManyRedirects`	重定向次数过多	超过最大重定向次数限制	警告
`SSLError`	SSL证书验证失败	证书无效、过期或不匹配	严重
`ProxyError`	代理服务器错误	代理服务器连接失败或无响应	警告
`ContentDecodingError`	内容解码失败	无法解码响应内容（如gzip损坏）	一般
`URLRequired`	URL缺失	请求没有指定有效的URL	严重

异常处理示例

ConnectionError

try:
    response = requests.get('http://nonexistent-domain.com')
except requests.exceptions.ConnectionError as e:
    print(f"连接失败: {e}")
    # 可能的恢复措施：
    # 1. 检查网络连接
    # 2. 尝试备用服务器
    # 3. 使用本地缓存

Timeout

try:
    response = requests.get(
        'https://slow-api.com/data',
        timeout=5  # 5秒超时
    )
except requests.exceptions.Timeout as e:
    print(f"请求超时: {e}")
    # 可能的恢复措施：
    # 1. 增加超时时间
    # 2. 实现重试机制
    # 3. 使用更快的备用服务

HTTP错误处理

HTTP错误（状态码4xx和5xx）不会自动引发异常，除非你显式调用response.raise_for_status()。

HTTP错误处理示例

import requests

def fetch_data_with_http_error_handling(url):
    """处理HTTP错误的示例函数"""
    try:
        response = requests.get(url)

        # 检查HTTP状态码，如果是错误状态码则引发HTTPError
        response.raise_for_status()

        # 如果状态码是2xx，处理响应
        data = response.json()
        return data

    except requests.exceptions.HTTPError as http_err:
        # HTTP错误处理
        status_code = response.status_code

        if status_code == 404:
            print(f"资源未找到: {url}")
            # 处理404错误的逻辑
            return None

        elif status_code == 401:
            print("认证失败，需要重新登录")
            # 刷新token或重新认证
            return refresh_and_retry(url)

        elif status_code == 403:
            print("权限不足，访问被拒绝")
            # 检查权限或使用其他账号
            return None

        elif status_code == 429:
            print("请求过于频繁，被限流")
            # 实现退避重试
            time.sleep(5)  # 等待5秒
            return fetch_data_with_http_error_handling(url)

        elif 500 <= status_code < 600:
            print(f"服务器内部错误: {status_code}")
            # 服务器错误，可以重试
            return retry_with_backoff(url)

        else:
            print(f"HTTP错误 {status_code}: {http_err}")
            return None

    except requests.exceptions.RequestException as req_err:
        # 其他Requests相关错误
        print(f"请求失败: {req_err}")
        return None

    except Exception as e:
        # 其他未预期的错误
        print(f"未预期的错误: {e}")
        return None

# 使用示例
if __name__ == "__main__":
    result = fetch_data_with_http_error_handling("https://api.example.com/data")
    if result:
        print("数据获取成功")
    else:
        print("数据获取失败")

raise_for_status()方法

response.raise_for_status() 方法会在HTTP状态码为4xx或5xx时引发HTTPError异常。

这是处理HTTP错误的推荐方式，因为它让你能够：

区分成功的响应和错误的响应
根据具体的状态码采取不同的恢复措施
避免在后续代码中处理错误数据

状态码分类

1xx (信息): 临时响应，通常不需要特殊处理
2xx (成功): 请求成功，可以安全处理响应
3xx (重定向): 需要进一步操作，Requests会自动处理
4xx (客户端错误): 请求有问题，需要修正
5xx (服务器错误): 服务器故障，可以稍后重试

超时处理

超时是网络请求中最常见的异常之一。Requests允许你设置两种超时：连接超时和读取超时。

超时参数详解

连接超时 (connect timeout)

建立TCP连接的最大等待时间。如果在这个时间内无法建立连接，会引发ConnectTimeout异常。

# 设置3秒连接超时
requests.get(url, timeout=(3.05, None))

读取超时 (read timeout)

从服务器接收数据的最大等待时间。如果服务器在这个时间内没有发送数据，会引发ReadTimeout异常。

# 设置10秒读取超时
requests.get(url, timeout=(None, 10))

完整的超时处理示例

import requests
import time

def make_request_with_timeout(url, max_retries=3):
    """带有超时处理和重试机制的请求函数"""

    # 超时配置：连接超时3秒，读取超时10秒
    timeout_config = (3.05, 10)

    for attempt in range(max_retries):
        try:
            print(f"第 {attempt + 1} 次尝试...")

            response = requests.get(url, timeout=timeout_config)
            response.raise_for_status()

            print("请求成功!")
            return response

        except requests.exceptions.ConnectTimeout:
            print(f"连接超时 - 服务器 {url} 无法连接")

            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # 指数退避：1, 2, 4秒
                print(f"等待 {wait_time} 秒后重试...")
                time.sleep(wait_time)
            else:
                print("已达到最大重试次数")
                return None

        except requests.exceptions.ReadTimeout:
            print(f"读取超时 - 服务器响应太慢")

            if attempt < max_retries - 1:
                # 对于读取超时，可以尝试增加读取超时时间
                new_timeout = (timeout_config[0], timeout_config[1] * 2)
                timeout_config = new_timeout
                print(f"增加读取超时到 {new_timeout[1]} 秒")
            else:
                print("已达到最大重试次数")
                return None

        except requests.exceptions.Timeout:
            # 通用超时处理（如果不知道是连接还是读取超时）
            print(f"请求超时")

            if attempt < max_retries - 1:
                wait_time = 1 * (attempt + 1)  # 线性退避
                print(f"等待 {wait_time} 秒后重试...")
                time.sleep(wait_time)
            else:
                print("已达到最大重试次数")
                return None

        except requests.exceptions.RequestException as e:
            print(f"请求失败: {e}")
            return None

    return None

# 使用示例
if __name__ == "__main__":
    result = make_request_with_timeout("https://httpbin.org/delay/5", max_retries=3)
    if result:
        print(f"响应状态码: {result.status_code}")
    else:
        print("所有尝试都失败了")

超时设置的最佳实践

始终设置超时：避免请求永远挂起
区分连接超时和读取超时：可以采取不同的恢复策略
根据场景调整超时值：API请求可以短一些，文件下载需要长一些
实现指数退避重试：对于超时错误，使用逐渐增加的重试间隔

重试机制

对于临时性错误（如网络波动、服务器重启），重试机制可以显著提高请求的成功率。

实现重试机制的三种方法

1. 手动实现

使用循环和条件判断手动实现重试逻辑，最灵活但代码较多。

2. 使用retrying库

使用第三方库简化重试逻辑，功能强大但需要额外依赖。

3. 使用urllib3的Retry

使用Requests底层库的功能，最原生但配置稍复杂。

方法一：手动实现重试机制

手动重试机制实现

import requests
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def make_request_with_retry(url, max_retries=3, backoff_factor=1):
    """
    带有重试机制的请求函数

    参数:
        url: 请求URL
        max_retries: 最大重试次数
        backoff_factor: 退避因子，用于计算重试间隔
    """
    # 需要重试的异常类型
    retry_exceptions = (
        requests.exceptions.ConnectionError,
        requests.exceptions.Timeout,
        requests.exceptions.HTTPError  # 只重试某些HTTP错误
    )

    # 需要重试的HTTP状态码
    retry_status_codes = {408, 429, 500, 502, 503, 504}

    for attempt in range(max_retries):
        try:
            logger.info(f"尝试 {attempt + 1}/{max_retries}: {url}")

            response = requests.get(url, timeout=(3.05, 10))

            # 检查是否需要重试（基于状态码）
            if response.status_code in retry_status_codes:
                response.raise_for_status()  # 这会引发HTTPError

            # 请求成功
            logger.info(f"请求成功，状态码: {response.status_code}")
            return response

        except retry_exceptions as e:
            logger.warning(f"请求失败: {type(e).__name__} - {e}")

            # 如果是最后一次尝试，直接抛出异常
            if attempt == max_retries - 1:
                logger.error(f"所有 {max_retries} 次尝试都失败了")
                raise

            # 计算等待时间（指数退避）
            wait_time = backoff_factor * (2 ** attempt)
            logger.info(f"等待 {wait_time} 秒后重试...")
            time.sleep(wait_time)

        except Exception as e:
            # 其他异常不重试
            logger.error(f"非重试异常: {type(e).__name__} - {e}")
            raise

    # 理论上不会执行到这里
    return None

# 使用示例
if __name__ == "__main__":
    try:
        response = make_request_with_retry(
            "https://httpbin.org/status/500",  # 模拟服务器错误
            max_retries=3,
            backoff_factor=1
        )
        if response:
            print("最终请求成功!")
    except Exception as e:
        print(f"所有重试都失败了: {e}")

方法二：使用requests.adapters.Retry

使用urllib3的Retry功能

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 创建重试策略
retry_strategy = Retry(
    total=3,  # 总重试次数（包括第一次请求）
    backoff_factor=1,  # 退避因子
    status_forcelist=[429, 500, 502, 503, 504],  # 需要重试的状态码
    allowed_methods=["GET", "POST"],  # 允许重试的HTTP方法
    raise_on_status=False  # 是否在状态码错误时引发异常
)

# 创建HTTP适配器并设置重试策略
adapter = HTTPAdapter(max_retries=retry_strategy)

# 创建会话并挂载适配器
session = requests.Session()
session.mount("http://", adapter)
session.mount("https://", adapter)

# 使用带有重试机制的会话发送请求
try:
    response = session.get("https://httpbin.org/status/500", timeout=5)
    print(f"状态码: {response.status_code}")
    print(f"重试次数: {response.raw.retries.total if response.raw else 0}")
except requests.exceptions.RequestException as e:
    print(f"请求失败: {e}")

# 也可以为特定请求自定义重试策略
custom_retry = Retry(
    total=5,
    backoff_factor=0.5,
    status_forcelist=[500, 502, 503, 504]
)

custom_adapter = HTTPAdapter(max_retries=custom_retry)
custom_session = requests.Session()
custom_session.mount("https://api.example.com", custom_adapter)

错误处理最佳实践

防御性编程

始终假设网络请求可能失败
验证输入URL的格式和有效性
检查响应数据的完整性和格式
为外部依赖设置合理的超时

分层错误处理

在底层处理技术性错误（连接、超时）
在业务层处理逻辑错误（验证、业务规则）
在展示层处理用户友好的错误消息
使用适当的日志级别记录错误

智能重试策略

区分可重试错误和不可重试错误
实现指数退避避免加重服务器负担
设置最大重试次数防止无限循环
考虑使用随机化退避（jitter）

监控和告警

监控请求成功率、延迟和错误率
为关键错误设置告警阈值
记录详细的错误上下文信息
实现错误仪表板和报告

[INFO] 开始请求: https://api.example.com/data

[WARNING] 连接超时，第1次重试...

[INFO] 等待2秒后重试

[SUCCESS] 请求成功! 状态码: 200

[INFO] 响应时间: 1.2秒

[INFO] 数据验证通过

错误处理检查清单

✓ 是否所有网络请求都有异常处理？
✓ 是否设置了合理的超时时间？
✓ 是否实现了重试机制？
✓ 是否记录了足够的信息用于调试？
✓ 是否向用户显示了友好的错误信息？
✓ 是否有降级方案或备用数据源？

实际应用示例

生产环境级别的API客户端

import requests
import time
import logging
from typing import Optional, Dict, Any
from dataclasses import dataclass
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

@dataclass
class ApiClientConfig:
    """API客户端配置"""
    base_url: str
    timeout: tuple = (3.05, 30)
    max_retries: int = 3
    retry_backoff: float = 1.0
    rate_limit_delay: float = 1.0

class ApiClientError(Exception):
    """自定义API客户端异常"""
    pass

class ApiClient:
    """生产环境级别的API客户端"""

    def __init__(self, config: ApiClientConfig):
        self.config = config
        self.session = self._create_session()
        self.last_request_time = 0

    def _create_session(self) -> requests.Session:
        """创建配置好的HTTP会话"""
        session = requests.Session()

        # 配置重试策略
        retry_strategy = Retry(
            total=self.config.max_retries,
            backoff_factor=self.config.retry_backoff,
            status_forcelist=[408, 429, 500, 502, 503, 504],
            allowed_methods=["GET", "POST", "PUT", "DELETE"],
            raise_on_status=False
        )

        # 创建适配器
        adapter = HTTPAdapter(max_retries=retry_strategy)

        # 挂载适配器
        session.mount("http://", adapter)
        session.mount("https://", adapter)

        # 设置默认请求头
        session.headers.update({
            'User-Agent': 'MyApiClient/1.0',
            'Accept': 'application/json',
            'Content-Type': 'application/json'
        })

        return session

    def _rate_limit(self):
        """简单的速率限制"""
        current_time = time.time()
        time_since_last = current_time - self.last_request_time

        if time_since_last < self.config.rate_limit_delay:
            sleep_time = self.config.rate_limit_delay - time_since_last
            logger.debug(f"速率限制，等待 {sleep_time:.2f} 秒")
            time.sleep(sleep_time)

        self.last_request_time = time.time()

    def _handle_error(self, response: Optional[requests.Response], error: Exception) -> None:
        """统一错误处理"""
        if response is not None:
            logger.error(
                f"HTTP错误: {response.status_code} - {response.reason} - URL: {response.url}"
            )

            # 尝试记录响应体中的错误信息
            try:
                error_data = response.json()
                logger.error(f"错误详情: {error_data}")
            except:
                logger.error(f"响应内容: {response.text[:500]}")

        # 根据异常类型采取不同措施
        if isinstance(error, requests.exceptions.Timeout):
            logger.error("请求超时，请检查网络连接或增加超时时间")
            raise ApiClientError("请求超时，请稍后重试") from error

        elif isinstance(error, requests.exceptions.ConnectionError):
            logger.error("网络连接失败，请检查网络设置")
            raise ApiClientError("网络连接失败，请检查网络") from error

        elif isinstance(error, requests.exceptions.HTTPError):
            # HTTP错误已在上面处理
            raise ApiClientError(f"服务器错误: {response.status_code}") from error

        else:
            logger.error(f"未预期的错误: {type(error).__name__} - {error}")
            raise ApiClientError("请求失败，请稍后重试") from error

    def request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
        """发送HTTP请求"""
        url = f"{self.config.base_url}{endpoint}"

        # 应用速率限制
        self._rate_limit()

        logger.info(f"发送请求: {method} {url}")

        try:
            # 发送请求
            response = self.session.request(
                method=method,
                url=url,
                timeout=self.config.timeout,
                **kwargs
            )

            # 记录请求统计
            logger.debug(
                f"请求完成 - 状态码: {response.status_code} "
                f"耗时: {response.elapsed.total_seconds():.2f}s"
            )

            # 检查HTTP状态码
            response.raise_for_status()

            # 解析响应
            try:
                return response.json()
            except ValueError as e:
                logger.warning(f"响应不是有效的JSON: {e}")
                return {"text": response.text}

        except requests.exceptions.RequestException as e:
            self._handle_error(response if 'response' in locals() else None, e)
            raise  # 这行实际上不会执行，因为_handle_error已经raise了

    def get(self, endpoint: str, **kwargs) -> Dict[str, Any]:
        """发送GET请求"""
        return self.request('GET', endpoint, **kwargs)

    def post(self, endpoint: str, data: Optional[Dict] = None, **kwargs) -> Dict[str, Any]:
        """发送POST请求"""
        return self.request('POST', endpoint, json=data, **kwargs)

    def close(self):
        """关闭会话"""
        self.session.close()
        logger.info("API客户端已关闭")

# 使用示例
if __name__ == "__main__":
    # 配置客户端
    config = ApiClientConfig(
        base_url="https://api.example.com/v1",
        timeout=(5, 30),
        max_retries=3,
        retry_backoff=1.0
    )

    # 创建客户端
    client = ApiClient(config)

    try:
        # 发送请求
        data = client.get("/users/123")
        print(f"获取的用户数据: {data}")

        # 发送POST请求
        new_user = {"name": "John Doe", "email": "john@example.com"}
        result = client.post("/users", data=new_user)
        print(f"创建用户结果: {result}")

    except ApiClientError as e:
        print(f"API请求失败: {e}")

    except KeyboardInterrupt:
        print("\n用户中断操作")

    finally:
        # 确保关闭客户端
        client.close()
        print("程序结束")

Python Requests库

Requests 错误与异常处理

本页内容