requests库中的那些高级用法

收录于 python

2020-03-12 约 2734 字预计阅读 6 分钟次阅读

作为python中最通用的http工具库之一——requests，具有简洁、直观和全面的特点，一般来说，requests与python中的内置模块urllib3已经逐渐成为程序员处理HTTP请求的首选工具。

得益于requests简洁的API接口，在生产环境下得到了非常普遍的应用，即使在复杂的应用场景下，request的扩展性也非常好。如果你正在写一个API客户端或者网络爬虫，同时对断网情况比较棘手，那么下面所介绍的requests高级技巧可能会帮助你更好的进行程序调试。

请求钩子

应用第三方API时，需要验证应答内容是否合法，requests提供了raise_for_status()方法来判断应答内容的HTTP状态码是不是4xx或者5xx，表明请求产生了客户端或者服务器错误。

例如：

1
2
3


response = requests.get('https://api.github.com/user/repos?page=1')
# 判断是否有无错误
response.raise_for_status()

每一次请求都调用raise_for_status会非常繁琐，requests非常贴心的提供了一个’钩子(hook)‘接口（通过在请求过程的特定部分指定回调函数）。下面请看案例代码，通过使用hook来确保每次服务应答后，raise_for_status能被调用。

1
2
3
4
5


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

设置基链接

假设你只使用api.org上的一个api，你可能在每次调用时都要重复编写http协议和域名。

1
2


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

使用BaseUrlSession可以避免敲下这些重复的内容。下面看案例代码，

1
2
3
4


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

！注意requests_toolbelt没有默认包含在requests中的，使用时是需要额外安装的

设置默认超时

requests文档中推荐在生产环境下设置超时，如果你忘记设置超时，应用程序可能会当掉，尤其是在同步环境下，

1

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

但是每次设置超时时间会非常繁琐，偶尔忘记设置超时会非常恼火。

使用Transport Adapters可以为所有的HTTP调用设置默认超时时间，当然使用后也是可以通过再定义覆盖默认配置，下面看案例代码，

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30


from requests.adapters import HTTPAdapter 
DEFAULT_TIMEOUT = 5 
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 = request.Session()
adapter = TimeoutHTTPAdapter(timeout=2.5) 
http.mount("https://",adapter)
http.mount("http://",adapter) 

#使用默认配置
response = http.get('https://api.weibo.com')

#覆盖默认配置
response = http.get('https://api.weibo.com',timeout=10)
   

失败后的重试

服务器当掉后，网络连接会变得拥塞或者有损，如果想要建立更具鲁棒性的系统，那么必须考虑网络连接失败，以及建立重试策略。在Http客户端上添加重试策略是非常直接的，我们来创建一个HTTPAdapter，然后在adapter上添加策略

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15


from requests.adapter import 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 类提供了健全的默认值，但是是高度可配置的，所以这里是我使用的最常见参数的纲要。其中

1

total=3

total代表重试的总次数，如果失败的请求或者重定向的次数超过这个数字，客户端将抛出 urllib3.exceptions.Maxretryerror 异常。通常3次重试就足够了。

1

status_forcelist=[413, 429, 503]

要重试的 HTTP 响应代码。您可能希望对常见的服务器错误(500、502、503、504)进行重试，因为服务器和反向代理并不总是遵循 HTTP 规范。总是在超出429速率限制的情况下重试，因为默认情况下，urllib 库应该在失败请求时增量地退出。

1

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

要重试的 HTTP 方法。默认情况下，这包括除 POST 以外的所有 HTTP 方法，因为 POST 可能导致新的插入。修改此参数以包含 POST，因为大多数 i 处理的 API 不返回错误代码并在同一调用中执行插入操作。如果他们这样做了，您可能应该发布一个 bug 报告。

1

backoff_factor=0

backoff_factor(退避因子)与失败的请求之间休眠的时间有关，其算法如下：

1

{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，表示不会设置截断二进制指数退避算法，重试将立即执行。

结合超时和重试

由于 HTTPAdapter 具有类比的特性，我们可以像下面这样将重试和超时结合起来:

1
2


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

调试HTTP请求

有时候请求会失败，而你却不知道为什么。记录请求和响应可以帮助您了解故障。有两种方法可以做到这一点——要么使用内置的调试日志记录设置，要么使用请求钩子。

打印HTTP头文件

更改大于0的日志记录调试级别将记录响应 HTTP 报头。这是最简单的选项，但它不允许您查看 HTTP 请求或响应体。如果您处理的 API 返回一个不适合日志记录或包含二进制内容的大体有效负载，那么它是有用的。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


import requests
import http

http.client.HTTPConnection.debuglevel = 1

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

# Output
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 生命周期，包括请求和响应的文本表示，那么您可以使用请求钩子和请求工具自带的转储组件。在处理基于 REST 的 API 时，我更喜欢这个选项，因为它不会返回非常大的响应。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41



import requests
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"})

# Output
< 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
      }
   ]
}

测试和模拟请求

在开发中使用第三方 API 会引入一个痛点——很难进行单元测试。为减轻这种痛苦，Sentry 的工程师在开发过程中编写了一个模拟请求的库。

不是将 HTTP 响应发送给服务器 getsentry / responses，而是截取 HTTP 请求，在应答时，测试过程中添加预定义的响应内容。

下面请看案例代码

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22


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)

如果发出的 HTTP 请求与模拟响应不匹配，则会抛出 ConnectionError。

1
2
3
4
5


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")

输出：

1
2
3
4
5
6
7


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

模仿浏览器行为

如果你已经写了足够多的网络爬虫代码，你将会注意到某些网站会根据浏览器或者请求方式返回不同的 HTML内容。有时这是一种反抓取措施，但通常服务器会进行用户代理嗅探，以找出最适合设备的内容(例如桌面或移动设备)。如果你想返回与浏览器显示的内容相同的内容，你可以使用 Firefox 或 Chrome 发送的内容覆盖默认的 User-Agent 头请求集。

1
2
3
4
5


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"
})

参考：

Advanced usage of Python requests - timeouts, retries, hooks

目录