十九、httpx 库

一. httpx 基础

HTTPX 是 Python 3 的全功能 HTTP 客户端，它提供同步和异步 API，并支持 HTTP/1.1 和 HTTP/2。

1.1 安装

1
pip install httpx

1.2 快速开始

1.2.1 get请求

1
2
3
4
5
6
7
import httpx

params = {
    "wd": "python"  # 输入百度搜索的内容
}
resp = httpx.get("https://www.baidu.com/s", params=params)  # 和原来requests的使用方法类似
print(resp.text)  # 获取数据信息

1.2.2 post请求

表单

1
2
3
4
5
import httpx

data = {'key1': 'value1', 'key2': 'value2'}
r = httpx.post("https://httpbin.org/post", data=data)
print(r.text)

文件

1
2
3
4
5
6
7
import httpx

files = {'upload-file': open('a.jpg', 'rb')}
# 也可以通过元组来指定数据类型
# files = {'upload-file': ('report.xls', open('report.xls', 'rb'), 'application/vnd.ms-excel')}
r = httpx.post("https://httpbin.org/post", files=files)
print(r.text)

JSON

1
2
3
4
5
import httpx

data = {'integer': 123, 'boolean': True, 'list': ['a', 'b', 'c']}
r = httpx.post("https://httpbin.org/post", json=data)
print(r.text)

二进制

1
2
3
4
5
6
7
import httpx

content = b'Hello, world'
r = httpx.post("https://httpbin.org/post", content=content, headers={
    "Content-Type": "application/octet-stream",
})
print(r.text)

Content-Type在上传二进制数据时设置自定义标头

上传二进制数据时格式类型
  • 常见的媒体格式类型:
    • text/html ： HTML格式
    • text/plain ：纯文本格式
    • text/xml ： XML格式
    • image/gif ：gif图片格式
    • image/jpeg ：jpg图片格式
    • image/png：png图片格式
  • 以application开头的媒体格式类型:
    • application/xhtml+xml ：XHTML格式
    • application/xml： XML数据格式
    • application/atom+xml ：Atom XML聚合格式
    • application/json： JSON数据格式
    • application/pdf：pdf格式
    • application/msword ： Word文档格式
    • application/octet-stream ： 二进制流数据（如常见的文件下载）
    • application/x-www-form-urlencoded ： <form encType="">中默认的encType，form表单数据被编码为key/value格式发送到服务器（表单默认的提交数据的格式）
  • 另外一种常见的媒体格式是上传文件之时使用的:
    • multipart/form-data ： 需要在表单中进行文件上传时，就需要使用该格式

1.2.3 响应处理

1
2
3
4
5
6
7
import httpx

resp = httpx.request("GET", "https://www.baidu.com")
if resp.status_code == httpx.codes.OK:
    print(resp.text)  # 如果请求成功

print(resp.raise_for_status())  # 判断响应是否成功，成功返回None，失败则报错

1.2.4 流式响应

对于大型下载，您可能希望使用不会一次将整个响应主体加载到内存中的流式响应。

1
2
3
4
5
6
7
8
9
import httpx
 
with httpx.stream("GET", "https://www.example.com") as r:
    for data in r.iter_bytes():  # 流式传输响应的二进制内容
        # for text in r.iter_text():  # 获取全部的文本内容
        # for line in r.iter_lines():  # 逐行获取传输响应的文本内容
        # for chunk in r.iter_raw():  # 获取编码前的原始数据
        # if r.headers['Content-Length'] < TOO_LONG:  # 有条件的加载内容
        print(data)

注意：
如果您以任何这些方式使用流式响应，则response.content and response.text属性将 不可用

1
2
3
4
5
6
7
8
9
10
11
12
13
import httpx
# 获取cookie
r = httpx.get('https://httpbin.org/cookies/set?chocolate=chip')
print(r.cookies['chocolate'])  # 获取请求中的cookie
 
# 设置cookie
cookies_1 = {"peanut": "butter"}
 
cookies_2 = httpx.Cookies()
cookies_2.set('cookie_on_domain', 'hello, there!', domain='httpbin.org')
cookies_2.set('cookie_off_domain', 'nope.', domain='example.org')
r = httpx.get('http://httpbin.org/cookies', cookies=cookies_2)
print(r.json())

1.2.6 重定向

默认情况下，HTTPX不会跟随所有 HTTP 方法的重定向，尽管这可以显式启用。
如，GitHub 将所有 HTTP 请求重定向到 HTTPS。

1
2
3
4
5
6
7
import httpx
r = httpx.get('http://github.com/')
print(r.status_code)
print(r.history)  # 查看重定向的记录
print(r.next_request)  # 获取到重定向以后的请求对象
resp = httpx.Client().send(r.next_request) # 对请求对象发送请求
print(resp.text)

follow_redirects=True : 跟随重定向

1
2
3
4
5
import httpx
r = httpx.get('http://github.com/', follow_redirects=True)
print(r.history)  # 查看重定向记录
print(r.url)  # 获取请求的url
print(r.text)  # 获取请求数据

1.2.7 超时和验证

接口请求默认超时为五秒, 可以通过 timeout 设置, timeout=None 表示禁用超时;

1
2
httpx.get('https://github.com/', timeout=0.001) 
httpx.get('https://github.com/', timeout=None) # 禁用超时

要提供基本身份验证凭据，请将纯文本str或bytes对象的 2 元组作为auth参数传递给请求函数：

1
2
3
4
5
6
import httpx

httpx.get("https://example.com", auth=("my_user", "password123"))  # 验证方法一

auth = httpx.DigestAuth("my_user", "password123")  # 验证方法二
httpx.get("https://example.com", auth=auth)

二、httpx 的两个坑

2.1 错误：httpx.ReadTimeout

  • 实测发现，网络不稳定的情况下，极其容易出现该错误。
  • 相对于 requests 库， httpx 库是有默认的超时时间的。
  • 参考方案： 初始化时将 timeout 赋值为 None
1
2
3
client = httpx.AsyncClient(timeout=None)
或者
httpx.get(url=url, timeout=None)

2.2 错误：SSL: CERTIFICATE_VERIFY_FAILED

  • 在多域名或者通配符域名证书的情况下，可能会出现此类错误。
  • 参考方案： 初始化时将 verify 赋值为 False
1
2
3
client = httpx.AsyncClient(verify=False)
 或者
httpx.get(url=url, verify=False)

2.3 综合

在引入 httpx 包 初始化时，将以上两个参数先行传入。

1
2
3
4
5
6
7
class NewClass:
    def __init__(self):
        self.client = httpx.AsyncClient(verify=False, timeout=None)
        _header = {
            "User-Agent": "Mozilla/5.0 (M1 Mac OS X 12) Safari/666.66"
        }
        self.client.headers.update(_header)

三、 客户端

3.1 特性

如果您来自 Requests，httpx.Client() 您可以使用它来代替 requests.Session().

1
2
3
4
5
6
7
8
9
10
# 使用方法1
with httpx.Client() as client:
    ...
     
# 使用方法2
client = httpx.Client()
try:
    ...
finally:
    client.close()

3.2 发出请求

一旦有了，就可以使用,等Client发送请求。例如：.get() .post() ，其传递参数的方法都一样，要注意一点的是，在实例化Client的时候，可以传入请求参数，使得这个局部作用域内可以共享这些参数，跨请求共享配置：

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
import httpx
 
# 共用请求头
url = 'http://httpbin.org/headers'
headers = {'user-agent': 'my-app/0.0.1'}
with httpx.Client(headers=headers) as client:
    # 这里面的所有请求的请求头都包含{'user-agent': 'my-app/0.0.1'}
    r = client.get(url)
 
print(r.json()['headers']['User-Agent'])
 
# 共用 + 私有
headers = {'X-Auth': 'from-client'}
params = {'client_id': 'client1'}
with httpx.Client(headers=headers, params=params) as client:
    headers_ = {'X-Custom': 'from-request'}
    params_ = {'request_id': 'request1'}
    r = client.get('https://example.com', headers=headers_,
                   params=params_)  # 这个参数结合了headers+headers_ , params+params_，但是只限于params和headers，对于所有其他参数，内部请求级别的值优先
 
print(r.request.url)
print(r.request.headers['X-Auth'])
print(r.request.headers['X-Custom'])
 
# 优先级
with httpx.Client(auth=('tom', 'mot123')) as client:
    r = client.get('https://example.com', auth=('alice', 'ecila123'))
 
_, _, auth = r.request.headers['Authorization'].partition(' ')
import base64
 
print(base64.b64decode(auth))

3.3、 其他配置

base_url 允许您为所有传出请求添加 URL

1
2
3
4
5
6
import httpx
 
with httpx.Client(base_url='http://httpbin.org') as client:
     r = client.get('/headers')
 
print(r.request.url)

3.4 python_web

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from flask import Flask
import httpx
 
app = Flask(__name__)
 
@app.route("/")
def hello():
    return "Hello World!"
 
with httpx.Client(app=app, base_url="http://localhost") as client:
    # base_url：指定app的根路由
    r = client.get("/")  # 获取根路由下的响应数据
    print(r.text)
    assert r.status_code == 200  # 断言
    assert r.text == "Hello World!"

3.5 Request对象

HTTPX 支持构建显式Request实例：

1
request = httpx.Request("GET", "https://example.com")

要将Request实例分派到网络，请创建一个Client实例并使用.send()：

1
2
3
with httpx.Client() as client:
    response = client.send(request)
    ...

2.6 进度条

监控下载进度，使用响应流并检查 response.num_bytes_downloaded 属性。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import tempfile
import httpx
from tqdm import tqdm
 
with tempfile.NamedTemporaryFile() as download_file:  # 创建一个临时文件。程序结束就删除
    url = "https://speed.hetzner.de/100MB.bin"
    with httpx.stream("GET", url) as response:  # 使用流发送请求
        total = int(response.headers["Content-Length"])
 
        with tqdm(total=total, unit_scale=True, unit_divisor=1024, unit="B") as progress:
            num_bytes_downloaded = response.num_bytes_downloaded
            for chunk in response.iter_bytes():
                download_file.write(chunk)
                progress.update(response.num_bytes_downloaded - num_bytes_downloaded)
                num_bytes_downloaded = response.num_bytes_downloaded