Python requests 高级用法：timeouts、retries、hooks

python-requests

原文地址：findwork.dev
作者：Dani Hodovic

在我使用的所有编程语言中，Python 的HTTP 库 requests 是我最喜欢的HTTP 工具。它简单，直观，在Python 社区中无处不在。大多数与HTTP 交互的程序要么使用它，要么使用标准库的urllib3。

由于简单的API ，requests 非常容易上手，不仅如此，在高级使用场景，它也有非常强的可扩展性。如果你在编写一个频繁调用API的客户端，或者一个网络爬虫，你需要能容忍网络异常，需要有效的debug 跟踪信息，还需要一些语法糖。

以下是我在编写大量调用JSON API 的网络爬虫或者程序时，从requests 中发掘出的一些有用的功能特性汇总。

请求钩子
设置BaseUrl
设置默认超时
失败时重试
结合超时和重试
调试 HTTP 请求
打印HTTP Header
打印所有信息
测试和伪装请求
模仿浏览器行为

请求钩子

在使用第三方API接口的时候，你通常都想要验证返回响应是否有效。Requests 提供了一个快捷的方法 raise_for_status()，它确保响应的HTTP 状态码不是4xx 或者5xx，即请求未触发客户端或服务器错误。

举个例子：

response = requests.get('https://api.github.com/user/repos?page=1')
# 确保此处无错误
response.raise_for_status()

每次请求都需要调用raise_for_status()，太繁复了。幸运的是，requests 库提供了一个类似“钩子”的接口，你可以在请求的各个阶段添加回调。

我们可以用钩子来确保每个响应对象都调用raise_for_status()。

# 创建一个自定义的requests 对象，修改全局模块抛出异常

http = requests.Session()

assert_status_hook = lambda response, *args, **kwargs: response.raise_for_status()
http.hooks["response"] = [assert_status_hook]

http.get("https://api.github.com/user/repos?page=1")

> HTTPError: 401 Client Error: Unauthorized for url: https://api.github.com/user/repos?page=1

设置BaseUrl

假设你仅使用api.org 上的API。每次http 调用，你都需要重复写协议和域名：

requests.get('https://api.org/list/')
requests.get('https://api.org/list/3/item')

你可以通过使用BaseUrlSession 少打一些字。它允许你为HTTP 客户端指定基础的url，这样每次请求时候只需要指定资源路径。

from requests_toolbelt import sessions
http = sessions.BaseUrlSession(base_url="https://api.org")
http.get("/list")
http.get("/list/item")

注意默认安装的requests 不包含requests toolbelt，你需要单独安装它。

设置默认超时

requests 文档建议你在所有生产环境的代码中都设置超时。如果忘记设置超时，异常的外部服务会导致你的应用hang 住，尤其大多数的Python 代码是同步的。

requests.get('https://github.com/', timeout=0.001)

但是这样做非常重复，容易遗漏。未来某一天当你发现是由于某人忘记设置超时导致生产环境的程序停止时，你会大发雷霆。

giphy

我们可以使用Transport Adapters 为每个HTTP 请求设置默认超时时间。它可以确保即使开发人员忘记在他的调用中设置 timeout=1 参数，也设置了一个合理的超时时间，并且每次调用都可以覆盖它。

下文是一个配置了默认超时的自定义 Transport Adapter 例子，灵感来自这条Github 评论。在构建http 客户端和send() 方法时，我们重写了构造函数，提供了一个默认的超时时间，以确保当未传递timeout 参数时会使用默认的超时时。

from requests.adapters import HTTPAdapter

DEFAULT_TIMEOUT = 5 # seconds

class TimeoutHTTPAdapter(HTTPAdapter):
    def __init__(self, *args, **kwargs):
        self.timeout = DEFAULT_TIMEOUT
        if "timeout" in kwargs:
            self.timeout = kwargs["timeout"]
            del kwargs["timeout"]
        super().__init__(*args, **kwargs)

    def send(self, request, **kwargs):
        timeout = kwargs.get("timeout")
        if timeout is None:
            kwargs["timeout"] = self.timeout
        return super().send(request, **kwargs)

可以这样使用它：

import requests

http = requests.Session()

# 挂载到 http、https 
adapter = TimeoutHTTPAdapter(timeout=2.5)
http.mount("https://", adapter)
http.mount("http://", adapter)

# 使用默认超时： 2.5s 
response = http.get("https://api.twilio.com/")

# 为特定请求覆盖超时时间
response = http.get("https://api.twilio.com/", timeout=10)

失败时重试

我们常常会遇到网络连接中断，拥塞，或者服务器故障。如果想编写一个强大的软件，我们必须考虑到失败的情况，并指定重试策略。

给HTTP 客户端添加重试策略非常简单。我们可以创建一个 HTTPAdapter，传入我们的重试策略。

from requests.packages.urllib3.util.retry import Retry

retry_strategy = Retry(
    total=3,
    status_forcelist=[429, 500, 502, 503, 504],
    method_whitelist=["HEAD", "GET", "OPTIONS"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
http = requests.Session()
http.mount("https://", adapter)
http.mount("http://", adapter)

response = http.get("https://en.wikipedia.org/w/api.php")

默认的Retry 类已经有合理的默认值，不过它是深度可定制化的，以下是一些我常用到的参数。

以下参数也包含requests 库的默认参数。

total=10

重试总次数。如果失败的请求数或重定向次数达到该值，客户端会抛出urllib3.exceptions.MaxRetryError 异常。我会根据我调用的API 来调整这个参数，不过我通常将其设置为小于10，一般3次重试就足够了。

status_forcelist=[413, 429, 503]

重试的HTTP 响应状态码。你可能想对常见的服务器错误（500，502，503，504） 都进行重试，因为服务器或者反向代理不遵循HTTP 规范。429 超出频率限制时总重试的原因是，urllib 库对于失败请求会有渐进性的退避策略，所以无碍。

method_whitelist=["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE"]

重试的HTTP 方法。默认情况下，该参数包含除了POST 之外的所有HTTP 方法，因为POST 会导致新增。修改该参数增加POST 是因为大多数我接触过的API 都都不会返回错误码，并且不会在相同的调用中执行新增。如果不是这样，你最好给他们提个bug。

backoff_factor=0

这个参数很有趣。它允许你修改进程在失败请求之间的休眠时间。算法是这样的：

{backoff factor} * (2 ** ({number of total retries} - 1))

举个例子，如果退避因子设置为：

1秒，休眠时间依次为0.5, 1, 2, 4, 8, 16, 32, 64, 128, 256
2秒 - 1, 2, 4, 8, 16, 32, 64, 128, 256, 512
10秒 - 5, 10, 20, 40, 80, 160, 320, 640, 1280, 2560

值呈指数增长，是重试策略的合理实现。

这个参数默认是0，意味着不会设置指数退避，重试会立即执行。确保它被设置为1，以免击垮你的服务器！。

重试模块的完整文档在这里。

结合超时和重试

因为HTTPAdapter 是一样的，我们可以把重试和超时像下面这样结合起来：

retries = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
http.mount("https://", TimeoutHTTPAdapter(max_retries=retries))

调试 HTTP 请求

有时请求失败，但是你无法找出原因。记录请求和响应或许能帮助你查明失败原因。有两种方法可以做到，使用内置的debug 日志配置，或者使用请求钩子。

打印HTTP Header

将日志调试等级调整到大于0，就会记录响应的HTTP Header。这是最简单的选项，但是它没法让你看到HTTP 请求和响应的包体。如果你调用的API会返回超大的包体或不合适打印出来，例如包含二进制内容，这个选项很有用。

任何大于0 的日志等级，都会开启调制日志。

import requests
import http

http.client.HTTPConnection.debuglevel = 1

requests.get("https://www.google.com/")

# 输出
send: b'GET / HTTP/1.1\r\nHost: www.google.com\r\nUser-Agent: python-requests/2.22.0\r\nAccept-Encoding: gzip, deflate\r\nAccept: */*\r\nConnection: keep-alive\r\n\r\n'
reply: 'HTTP/1.1 200 OK\r\n'
header: Date: Fri, 28 Feb 2020 12:13:26 GMT
header: Expires: -1
header: Cache-Control: private, max-age=0

打印所有信息

如果你想打印整个HTTP 生命周期，包括请求和响应内容的文本内容，你可以使用请求钩子和requests_toolbelt 中的dump 工具。

当我调用不会返回很多的响应内容的REST API 时，我更喜欢用这个选项。

from requests_toolbelt.utils import dump

def logging_hook(response, *args, **kwargs):
    data = dump.dump_all(response)
    print(data.decode('utf-8'))

http = requests.Session()
http.hooks["response"] = [logging_hook]

http.get("https://api.openaq.org/v1/cities", params={"country": "BA"})

# 参考
< GET /v1/cities?country=BA HTTP/1.1
< Host: api.openaq.org

> HTTP/1.1 200 OK
> Content-Type: application/json; charset=utf-8
> Transfer-Encoding: chunked
> Connection: keep-alive
>
{
   "meta":{
      "name":"openaq-api",
      "license":"CC BY 4.0",
      "website":"https://docs.openaq.org/",
      "page":1,
      "limit":100,
      "found":1
   },
   "results":[
      {
         "country":"BA",
         "name":"Goražde",
         "city":"Goražde",
         "count":70797,
         "locations":1
      }
   ]
}

参考 https://toolbelt.readthedocs.io/en/latest/dumputils.html

测试和伪装请求

使用第三方API 会给开发人员带来一个痛点，它们很难进行单元测试。Sentry 的工程师通过写了一个库来伪装请求，减轻了开发过程中的痛苦。

无需发送请求到服务器，getsentry/responses 截获了HTTP 请求并返回一个你的测试过程中添加的预先定义好的内容。

import unittest
import requests
import responses


class TestAPI(unittest.TestCase):
    @responses.activate  # intercept HTTP calls within this method
    def test_simple(self):
        response_data = {
                "id": "ch_1GH8so2eZvKYlo2CSMeAfRqt",
                "object": "charge",
                "customer": {"id": "cu_1GGwoc2eZvKYlo2CL2m31GRn", "object": "customer"},
            }
        # mock the Stripe API
        responses.add(
            responses.GET,
            "https://api.stripe.com/v1/charges",
            json=response_data,
        )

        response = requests.get("https://api.stripe.com/v1/charges")
        self.assertEqual(response.json(), response_data)

如果发出一个和伪装的响应不一致的请求，则会引发 ConnectionError。

class TestAPI(unittest.TestCase):
    @responses.activate
    def test_simple(self):
        responses.add(responses.GET, "https://api.stripe.com/v1/charges")
        response = requests.get("https://invalid-request.com")

输出

requests.exceptions.ConnectionError: Connection refused by Responses - the call doesn't match any registered mock.

Request:
- GET https://invalid-request.com/

Available matches:
- GET https://api.stripe.com/v1/charges

模仿浏览器行为

如果你编写过足够多的web 爬虫代码，你会注意到某些网站会针对你是通过浏览器还是代码访问会返回不同的HTML。有时这是一种防爬虫的策略，但是通常是由于服务器会居于User-Agent 嗅探，找出最适合访问设备的内容（例如桌面端或移动端）。

如果你想要服务器返回浏览器一样的内容，你可以重写请求的 User-Agent 头信息为FIrefox或Chrome。

import requests
http = requests.Session()
http.headers.update({
    "User-Agent": "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:68.0) Gecko/20100101 Firefox/68.0"
})

fangpsh's blog