在 Python 中管理请求失败

学习如何使用有效的重试策略和自定义逻辑来处理 Python 中失败的 HTTP 请求。
3 min read
在 Python 中管理请求失败 - 博文配图

当你处理 HTTP 请求时,请求失败是不可避免的现实,需要我们妥善应对。在 Web 开发中,状态码 200 表示一个良好的响应。然而,我们并不总能得到 200。本指南将帮助你理解如何处理那些非 200 状态码。

根据 Mozilla 的文档,状态码可分为以下几类:

  • 100-199:信息性响应
  • 200-299:成功响应
  • 300-399:重定向消息
  • 400-499:客户端错误消息
  • 500-599:服务器错误消息

什么是状态码?

错误码 很重要。在构建客户端程序(如网络爬虫)时,我们主要需要关注 400+ 和 500+ 范围内的状态码。400 段的错误通常是客户端方面的问题,比如身份验证、限流、超时,以及著名的 404:未找到文件 错误。500 段的错误则通常与服务器问题有关。

几十年来,Mozilla 一直在记录来自 W3CIETF 的 Web 开发标准。下面列出了一些你可能会遇到的常见错误码(并不完整)。这些错误来自 Mozilla 的 官方文档。根据你的目标站点,状态码可能略有不同,但逻辑应保持一致。

状态码 含义 描述
400 错误请求 检查请求格式
401 未授权 检查你的 API Key
403 禁止访问 你无法访问此数据
404 未找到 站点/端点不存在
408 请求超时 请求超时,请重试
429 请求过多 降低请求发送频率
500 服务器内部错误 通用服务器错误,请重试
501 未实现 服务器尚不支持此功能
502 错误网关 上游服务器返回失败响应
503 服务不可用 服务器暂时不可用,请稍后重试
504 网关超时 等待上游服务器超时

重试策略

如果想要实现重试机制,可以使用现成的库,例如 HTTPAdapter 和 Tenacity。根据你的需求,你甚至可以编写自己的重试逻辑。

通常,我们需要一个重试上限以及退避(backoff)策略。我们需要一个上限以避免陷入无限重试循环,同时需要逐步退避来避免给目标服务器造成过大压力。如果请求发送得太频繁,你会被屏蔽,或者服务器不堪负荷。

  • 重试上限:你需要设定一个上限。在尝试了 X 次之后,你的爬虫将放弃。
  • 退避算法:它相对简单。你希望从较短的等待时间开始,并在每次重试后递增。我们可从 0.3 开始,然后增加到 0.6、1.2 等等。

我们希望在一定次数内重试请求。每次请求失败后,我们都要等待更长一点的时间。

HTTPAdapter

使用 HTTPAdapter 时,我们需要配置三个主要内容:totalbackoff_factorstatus_forcelistallowed_methods 并不是必需,但它能帮助我们更好地定义重试条件,让代码更安全。下面的示例中,我们使用 httpbin 来自动触发错误并测试重试逻辑。

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

#create a session
session = requests.Session()

#configure retry settings
retry = Retry(
    total=3,                  #maximum retries
    backoff_factor=0.3,       #time between retries (exponential backoff)
    status_forcelist=(429, 500, 502, 503, 504), #status codes to trigger a retry
    allowed_methods={"GET", "POST"}
)

#mount the adapter with our custom settings
adapter = HTTPAdapter(max_retries=retry)
session.mount("http://", adapter)
session.mount("https://", adapter)

#actually make a request with our retry logic
try:
    print("Making a request with retry logic...")
    response = session.get("https://httpbin.org/status/500")
    response.raise_for_status()
    print("✅ Request successful:", response.status_code)
except requests.exceptions.RequestException as e:
    print("❌ Request failed after retries:", e)

在创建 Session 对象后,我们进行了如下操作:

  • 创建一个 Retry 对象,并定义如下内容:
    • total:请求的最大重试次数。
    • backoff_factor:重试间隔时间。随着重试次数的增加,这个值将指数级增加。
    • status_forcelist:一组“错误”状态码。任何位于此列表中的状态码都会自动触发重试。
  • 创建一个 HTTPAdapter 对象,并将其与 retry 绑定:adapter = HTTPAdapter(max_retries=retry)
  • 创建好 adapter 后,将其分别挂载到 HTTP 和 HTTPS 协议上:session.mount()

运行上述代码后,我们设置的 3 次重试(total=3)将会执行,最终输出会类似下面这样:

Making a request with retry logic...
❌ Request failed after retries: HTTPSConnectionPool(host='httpbin.org', port=443): Max retries exceeded with url: /status/500 (Caused by ResponseError('too many 500 error responses'))

Tenacity

你也可以使用 Tenacity 这个流行的开源 Python 重试库。它并不只限于 HTTP 请求,但能为我们提供一种更具可读性和可扩展性的方式来实现重试。

首先,你需要安装它:

pip install tenacity

安装完成后,我们可以创建一个装饰器,并用它来包装一个发送请求的函数。通过 @retry 装饰器,我们可添加 stopwaitretry 参数。

import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type, RetryError

#define a retry strategy
@retry(
    stop=stop_after_attempt(3),  #retry up to 3 times
    wait=wait_exponential(multiplier=0.3),  #exponential backoff
    retry=retry_if_exception_type(requests.exceptions.RequestException),  #retry on request failures
)

def make_request():
    print("Making a request with retry logic...")
    response = requests.get("https://httpbin.org/status/500")
    response.raise_for_status()
    print("✅ Request successful:", response.status_code)
    return response

# Attempt to make the request
try:
    make_request()
except RetryError as e:
    print("❌ Request failed after all retries:", e)

这里的逻辑和设置与我们在 HTTPAdapter 示例中的类似:

  • stop=stop_after_attempt(3):告诉 tenacity 在失败 3 次后放弃。
  • wait=wait_exponential(multiplier=0.3) 使用与之前相同的等待策略,也会进行指数级退避。
  • retry=retry_if_exception_type(requests.exceptions.RequestException) 告诉 tenacity 在每次出现 RequestException 异常时都进行重试。
  • make_request() 会发送请求至这个错误端点,并继承了装饰器中定义的所有特性。

运行此代码,你将看到类似的输出:

Making a request with retry logic...
Making a request with retry logic...
Making a request with retry logic...
❌ Request failed after all retries: RetryError[<Future at 0x75e762970760 state=finished raised HTTPError>]

自定义重试机制

你也可以自己编写重试机制,当你需要定制化的代码时,这通常是最好的选择。仅用相对少量的代码,就可实现与这些库相似的功能。

在下面的示例中,我们导入了 sleep 用来实现指数退避。我们同样定义了 totalbackoff_factorbad_codes。然后使用 while 循环来实现重试逻辑。只要还有剩余重试次数且尚未成功,我们就会尝试请求。

import requests
from time import sleep

#create a session
session = requests.Session()

#define our retry settings
total = 3
backoff_factor = 0.3
bad_codes = [429, 500, 502, 503, 504]

#try counter and success boolean
current_tries = 0
success = False

#attempt until we succeed or run out of tries
while current_tries < total and not success:
    try:
        print("Making a request with retry logic...")
        response = session.get("https://httpbin.org/status/500")
        if response.status_code in bad_codes:
            raise requests.exceptions.HTTPError(f"Received {response.status_code}, triggering retry")
        print("✅ Request successful:", response.status_code)
        success = True
    except requests.exceptions.RequestException as e:
        print(f"❌ Request failed: {e}, retries left: {total-current_tries}")
        sleep(backoff_factor)
        backoff_factor = backoff_factor * 2
        current_tries+=1

这里的核心逻辑由一个简单的 while 循环实现:

  • 如果 response.status_code 在我们的 bad_codes 列表中,我们就抛出异常。
  • 当某次请求失败时,我们会:
    • 打印错误信息到控制台。
    • sleep(backoff_factor) 用于在发出下一次请求前等待。
    • backoff_factor = backoff_factor * 2 会将退避因子加倍,为下一次重试做准备。
    • 递增 current_tries 以防止无限循环。

以下是我们自定义重试逻辑的输出示例:

Making a request with retry logic...
❌ Request failed: Received 500, triggering retry, retries left: 3
Making a request with retry logic...
❌ Request failed: Received 500, triggering retry, retries left: 2
Making a request with retry logic...
❌ Request failed: Received 500, triggering retry, retries left: 1

绕过封锁

在实际应用中,一些网站会对你进行封锁。最好的做法是始终 在 Python 的 requests 中使用代理。使用代理可以让你的请求从另一台机器发出,从而保护你的真实 IP,不会被目标站点屏蔽。我们还有一篇 详细指南,介绍如何绕过 IP 封锁。我们的 住宅代理 能帮助你应对这些挑战。

总结

现在你已经知道如何在 Python 中处理失败的 HTTP 请求了。无论你在写爬虫、API 客户端还是自动化工具,都能使用以上方法来解决这些问题。为了避免各种类型的请求失败,我们开发了像 Web Unlocker APIScraping Browser 这样的产品。这些工具能自动处理反爬虫措施、CAPTCHA 验证和 IP 封锁,保证在面对最具挑战性的网站时依旧能流畅且高效地进行网络爬取。

立即注册并开始免费试用吧。