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

每次运行爬虫脚本时，最让人头疼的莫过于那些意料之外的HTTP错误。你可能遇到过这样的情况：脚本运行了几个小时，最后却因为一个404错误而中断，所有数据都丢失了。更糟糕的是，你可能根本没有注意到这些错误，直到分析数据时才发现大量缺失的记录。

1. 为什么手动检查状态码是个糟糕的主意

在Python的requests库中，即使请求失败（比如返回了404或500状态码），默认情况下也不会抛出异常。这意味着开发者必须手动检查每个响应的状态码，这种做法看似简单，实则隐藏着诸多问题。

手动检查的典型代码示例 ：

response = requests.get('https://example.com/api')
if response.status_code == 200:
    # 处理成功响应
    data = response.json()
else:
    # 处理错误
    print(f"请求失败，状态码：{response.status_code}")

这种模式至少有三大缺陷：

1. 代码冗余 ：每个请求都需要重复相同的状态码检查逻辑
2. 错误处理不完整 ：容易忽略非200但仍有意义的状态码（如204 No Content）
3. 异常处理分离 ：网络连接错误和HTTP错误被分开处理，增加了复杂度

我曾经维护过一个爬虫项目，其中充斥着这样的代码片段。随着项目规模扩大，错误处理逻辑变得支离破碎，最终导致难以维护的代码库。

2. raise_for_status()的工作原理与优势

raise_for_status() 是requests库中一个简单但强大的方法，它会检查响应的HTTP状态码，如果状态码表示错误（4xx或5xx），则抛出 HTTPError 异常。

基本用法对比 ：

方法	代码量	可读性	错误处理一致性
手动检查	多	一般	差
raise_for_status()	少	好	好

实际应用示例 ：

try:
    response = requests.get('https://example.com/api')
    response.raise_for_status()  # 关键的一行
    data = response.json()
except requests.exceptions.HTTPError as err:
    print(f"HTTP错误发生: {err}")
except requests.exceptions.RequestException as err:
    print(f"请求异常: {err}")

这种方法将各种类型的错误统一到异常处理流程中，使代码更加清晰和健壮。我在重构前述项目时，通过引入 raise_for_status() ，将错误处理代码减少了约40%，同时显著提高了可靠性。

3. 构建生产级爬虫的错误处理策略

在实际爬虫项目中，单纯使用 raise_for_status() 还不够。我们需要结合其他技术构建完整的错误处理体系。

3.1 重试机制

网络请求可能会因为各种临时性问题失败，合理的重试策略可以大大提高爬虫的健壮性。

带指数退避的重试示例 ：

from time import sleep
from random import random

def make_request_with_retry(url, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.get(url, timeout=10)
            response.raise_for_status()
            return response
        except requests.exceptions.RequestException as err:
            if attempt == max_retries - 1:
                raise
            wait_time = (2 ** attempt) + random()
            print(f"请求失败，将在{wait_time:.1f}秒后重试...")
            sleep(wait_time)

3.2 全面的异常处理

不同的异常类型需要不同的处理方式。以下是一个更完整的异常处理框架：

try:
    response = requests.get(url, timeout=10)
    response.raise_for_status()
except requests.exceptions.HTTPError as err:
    # 处理HTTP错误（4xx, 5xx）
    log_error(f"HTTP错误 {err.response.status_code}: {url}")
    if err.response.status_code == 404:
        handle_missing_page(url)
    elif err.response.status_code == 429:
        handle_rate_limit()
except requests.exceptions.ConnectionError:
    # 处理连接问题
    log_error(f"连接失败: {url}")
except requests.exceptions.Timeout:
    # 处理超时
    log_error(f"请求超时: {url}")
except requests.exceptions.RequestException as err:
    # 处理其他requests异常
    log_error(f"请求异常: {err}")

3.3 日志记录

良好的日志记录对于调试和监控爬虫至关重要。以下是一个日志配置示例：

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('crawler.log'),
        logging.StreamHandler()
    ]
)

def log_error(message):
    logging.error(message)
    # 可以添加额外的错误跟踪逻辑

4. 高级应用场景与最佳实践

4.1 结合Session对象使用

对于需要多次请求的爬虫，使用Session对象可以提高性能并保持一致的配置。

with requests.Session() as session:
    session.headers.update({'User-Agent': 'MyCrawler/1.0'})
    try:
        response = session.get('https://example.com/api', timeout=5)
        response.raise_for_status()
        # 处理响应
    except requests.exceptions.RequestException as err:
        handle_error(err)

4.2 自定义异常处理

对于特定项目，可以创建自定义异常类来更好地封装错误处理逻辑。

class CrawlerError(Exception):
    """爬虫基础异常类"""
    pass

class PageNotFoundError(CrawlerError):
    """页面不存在异常"""
    pass

def make_request(url):
    try:
        response = requests.get(url)
        response.raise_for_status()
        return response
    except requests.exceptions.HTTPError as err:
        if err.response.status_code == 404:
            raise PageNotFoundError(f"页面不存在: {url}") from err
        raise CrawlerError(f"HTTP错误: {err}") from err

4.3 性能监控与警报

在生产环境中，监控爬虫的成功率和性能非常重要。

import time
from statistics import mean

class RequestMonitor:
    def __init__(self):
        self.request_times = []
        self.error_count = 0
    
    def track_request(self, func):
        def wrapper(*args, **kwargs):
            start = time.time()
            try:
                result = func(*args, **kwargs)
                self.request_times.append(time.time() - start)
                return result
            except Exception as err:
                self.error_count += 1
                raise
        return wrapper
    
    @property
    def avg_response_time(self):
        return mean(self.request_times) if self.request_times else 0
    
    @property
    def error_rate(self):
        total = len(self.request_times) + self.error_count
        return self.error_count / total if total else 0

5. 实战：构建健壮的爬虫框架

结合前面介绍的技术，我们可以构建一个完整的爬虫框架。以下是一个简化但功能完整的实现：

import requests
import logging
from time import sleep
from random import random
from functools import wraps

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

class RobustCrawler:
    def __init__(self, base_url, max_retries=3):
        self.base_url = base_url
        self.max_retries = max_retries
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'RobustCrawler/1.0',
            'Accept': 'application/json'
        })
    
    def request_with_retry(self, endpoint, method='GET', **kwargs):
        url = f"{self.base_url}{endpoint}"
        
        for attempt in range(self.max_retries):
            try:
                response = self.session.request(
                    method, 
                    url,
                    timeout=10,
                    **kwargs
                )
                response.raise_for_status()
                return response
            except requests.exceptions.HTTPError as err:
                logger.error(f"HTTP错误: {err}")
                if attempt == self.max_retries - 1:
                    raise
            except requests.exceptions.RequestException as err:
                logger.error(f"请求异常: {err}")
                if attempt == self.max_retries - 1:
                    raise
            
            wait_time = (2 ** attempt) + random()
            logger.info(f"将在{wait_time:.1f}秒后重试...")
            sleep(wait_time)
    
    def crawl(self, endpoint):
        try:
            response = self.request_with_retry(endpoint)
            return self.process_response(response)
        except requests.exceptions.HTTPError as err:
            logger.error(f"最终请求失败: {err}")
            return None
    
    def process_response(self, response):
        """子类应重写此方法以实现特定处理逻辑"""
        return response.json()

这个框架提供了重试机制、统一的错误处理和日志记录，可以作为更复杂爬虫项目的基础。在实际项目中，我发现这种结构显著减少了意外中断和数据丢失的情况。