19.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