目录

requests库中的那些高级用法

作为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) 

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

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

参考: