别再手动检查状态码了！用requests的raise_for_status()让你的Python爬虫更健壮

在编写Python爬虫或API调用脚本时，开发者最常遇到的挑战之一就是处理各种HTTP错误。想象一下这样的场景：你精心设计的爬虫运行了几个小时后突然崩溃，原因是一个简单的404错误未被捕获；或者你的数据分析流程因为一个未处理的429状态码而中断，导致整个批处理作业失败。这些问题不仅浪费时间，还可能造成数据丢失。

传统的手动检查状态码方法虽然直观，但往往导致代码冗长、可读性差，并且容易遗漏某些错误情况。requests库提供的 raise_for_status() 方法正是为解决这些问题而生。它能够自动检查HTTP响应状态码，并在遇到非2xx状态码时抛出异常，让错误处理变得更加结构化、高效。

1. 为什么需要更好的错误处理机制

网络请求本质上是不稳定的操作。服务器可能暂时不可用、资源可能被移动或删除、请求可能被限流——这些情况都会反映在HTTP状态码中。根据HTTP协议规范，状态码被分为几个主要类别：

• 2xx ：成功（如200 OK、201 Created）
• 3xx ：重定向（如301 Moved Permanently、302 Found）
• 4xx ：客户端错误（如404 Not Found、403 Forbidden）
• 5xx ：服务器错误（如500 Internal Server Error、503 Service Unavailable）

手动检查这些状态码的典型代码可能长这样：

response = requests.get('https://api.example.com/data')
if response.status_code == 200:
    process_data(response.json())
elif response.status_code == 404:
    log_error('Resource not found')
elif response.status_code == 500:
    log_error('Server error')
elif response.status_code == 429:
    wait_and_retry()
else:
    log_error(f'Unexpected status code: {response.status_code}')

这种方式的缺点显而易见：

1. 代码冗长 ：每个可能的状态码都需要单独处理
2. 可维护性差 ：新增状态码处理需要修改多处条件判断
3. 容易遗漏 ：开发者可能忘记检查某些重要状态码
4. 错误处理分散 ：难以集中管理所有网络请求相关的错误

raise_for_status() 方法通过将状态码检查标准化，有效解决了这些问题。

2. raise_for_status()的工作原理与基本用法

raise_for_status() 是requests.Response对象的一个方法，它会检查当前响应的状态码：

• 如果状态码在200-299范围内（表示成功），方法什么也不做
• 如果状态码不在这个范围内，则抛出 requests.exceptions.HTTPError 异常

基本使用模式如下：

import requests

try:
    response = requests.get('https://api.example.com/data')
    response.raise_for_status()  # 如果状态码不是2xx，这里会抛出异常
    data = response.json()
    process_data(data)
except requests.exceptions.HTTPError as http_err:
    print(f'HTTP错误发生: {http_err}')
except requests.exceptions.RequestException as req_err:
    print(f'请求异常: {req_err}')

这种结构的优势在于：

• 集中错误处理 ：所有网络请求错误都在同一个try-except块中处理
• 代码更简洁 ：不需要写多个if-elif来检查状态码
• 更安全 ：确保在继续处理响应数据前请求已成功
• 更易扩展 ：可以轻松添加更多异常类型处理

2.1 异常类型详解

requests库定义了多种异常类型来处理不同种类的请求错误：

异常类型	触发条件	典型场景
HTTPError	raise_for_status()检测到非2xx状态码	404, 500等HTTP错误
ConnectionError	无法建立连接	DNS解析失败，服务器拒绝连接
Timeout	请求超时	服务器响应过慢
TooManyRedirects	重定向次数过多	重定向循环
RequestException	所有requests异常的基类	捕获所有requests相关错误

合理利用这些异常类型可以构建更健壮的错误处理系统：

try:
    response = requests.get('https://api.example.com/data', timeout=5)
    response.raise_for_status()
    data = response.json()
except requests.exceptions.HTTPError as err:
    logger.error(f"HTTP错误: {err}")
    # 特殊处理404错误
    if response.status_code == 404:
        handle_not_found()
    elif response.status_code == 429:
        handle_rate_limit()
except requests.exceptions.Timeout:
    logger.error("请求超时")
    retry_later()
except requests.exceptions.ConnectionError:
    logger.error("连接错误")
    check_network()
except requests.exceptions.RequestException as err:
    logger.error(f"未知请求错误: {err}")

3. 实战：构建健壮的API客户端

让我们通过一个实际案例来展示如何利用 raise_for_status() 构建一个健壮的API客户端。假设我们需要从GitHub API获取用户的仓库信息。

3.1 基础实现

import requests
from requests.exceptions import HTTPError, RequestException
import time

def get_github_repos(username, retries=3, backoff_factor=1):
    url = f"https://api.github.com/users/{username}/repos"
    
    for attempt in range(retries):
        try:
            response = requests.get(url)
            response.raise_for_status()
            return response.json()
        except HTTPError as err:
            if response.status_code == 404:
                raise ValueError(f"用户 {username} 不存在") from err
            elif response.status_code == 403 and 'rate limit' in str(err):
                reset_time = int(response.headers.get('X-RateLimit-Reset', 0))
                wait_time = max(reset_time - time.time(), 0) + 10
                if attempt < retries - 1:
                    time.sleep(wait_time)
                    continue
            raise
        except RequestException as err:
            if attempt < retries - 1:
                time.sleep(backoff_factor * (attempt + 1))
                continue
            raise
    return None

这个实现包含了几项关键改进：

1. 自动重试机制 ：对于可重试的错误（如速率限制、临时网络问题），自动进行重试
2. 指数退避 ：每次重试等待时间逐渐增加，避免加重服务器负担
3. 特定错误处理 ：对404和403等常见错误进行特殊处理
4. 清晰的错误传播 ：将API特定的错误转换为更有意义的异常

3.2 高级技巧：创建自定义重试策略

对于生产环境的应用，我们可以使用 urllib3 的 Retry 类与 requests.Session 结合，实现更灵活的重试策略：

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

def create_retry_session(retries=3, backoff_factor=0.5, status_forcelist=(500, 502, 504)):
    session = requests.Session()
    retry = Retry(
        total=retries,
        read=retries,
        connect=retries,
        backoff_factor=backoff_factor,
        status_forcelist=status_forcelist,
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    return session

def get_with_retry(url):
    session = create_retry_session()
    try:
        response = session.get(url)
        response.raise_for_status()
        return response.json()
    except HTTPError as err:
        handle_http_error(err)
    except RequestException as err:
        handle_request_error(err)

这种方式的优势在于：

• 统一的重试策略 ：对所有请求应用相同的重试规则
• 更细粒度的控制 ：可以分别为连接错误、读取错误设置不同重试次数
• 自动处理 ：无需手动实现重试逻辑，代码更简洁

4. 最佳实践与常见陷阱

4.1 最佳实践

1. 始终检查响应状态 ：即使你认为请求应该成功，也要使用 raise_for_status()
2. 合理设置超时 ：避免请求无限期挂起

   requests.get(url, timeout=(3.05, 27))  # 连接超时3.05秒，读取超时27秒
   

3. 使用会话(Session) ：复用TCP连接提高性能

   with requests.Session() as session:
       session.get(url1)
       session.post(url2)
   

4. 记录完整的错误信息 ：包括URL、状态码、响应体等

   except HTTPError as err:
       logger.error(f"请求失败: {err.request.url} - {err.response.status_code} - {err.response.text}")
   

5. 考虑实现断路器模式 ：当错误率达到阈值时暂时停止请求

4.2 常见陷阱

1. 忽略响应内容 ：某些API在错误时也返回200状态码，但通过响应体表示错误

   data = response.json()
   if data.get('error'):
       raise ApiError(data['error'])
   

2. 过度依赖重试 ：对于非幂等操作（如POST请求），盲目重试可能导致重复操作
3. 不处理连接错误 ：只捕获HTTPError而忽略ConnectionError等
4. 泄露敏感信息 ：在错误日志中记录完整的API密钥或敏感数据
5. 不设置用户代理 ：某些API要求有效的User-Agent头

   headers = {'User-Agent': 'MyApp/1.0'}
   requests.get(url, headers=headers)
   

4.3 性能考虑

当处理大量请求时，错误处理的效率变得尤为重要。以下是一些优化建议：

1. 批量处理错误 ：对于批量请求，可以收集所有错误后统一处理
2. 异步请求 ：使用 aiohttp 或 httpx 进行并发请求

   import httpx
   
   async def fetch_url(url):
       async with httpx.AsyncClient() as client:
           try:
               response = await client.get(url)
               response.raise_for_status()
               return response.json()
           except httpx.HTTPStatusError as err:
               handle_error(err)
   

3. 缓存错误响应 ：对于暂时性错误，可以缓存并稍后重试
4. 监控错误率 ：跟踪不同端点的错误率，及时发现API问题

在实际项目中，我经常遇到需要同时处理数百个API请求的情况。使用 raise_for_status() 结合适当的错误处理策略，可以显著提高代码的可靠性和可维护性。一个常见的模式是创建自定义的API客户端类，封装所有的错误处理逻辑：

class ApiClient:
    def __init__(self, base_url):
        self.base_url = base_url
        self.session = create_retry_session()
    
    def get_resource(self, resource_id):
        url = f"{self.base_url}/resources/{resource_id}"
        try:
            response = self.session.get(url)
            response.raise_for_status()
            return response.json()
        except HTTPError as err:
            if err.response.status_code == 404:
                raise ResourceNotFound(f"Resource {resource_id} not found") from err
            raise ApiError(f"API request failed: {err}") from err
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.session.close()

这种封装使得业务代码可以更专注于核心逻辑，而不必担心底层的网络错误处理。