02.httpx 库

Httpx 官方文档 , 支持同步和异步 API，并支持 HTTP/1.1 和 HTTP/2

一. httpx 基础

提示

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

1.1 安装

pip install httpx

1.2 快速开始

1.2.1 get 请求

import httpx

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

1.2.2 post 请求

表单

import httpx

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

文件

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

import httpx

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

二进制

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 响应处理

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 流式响应

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

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.5 cookie

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。

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 : 跟随重定向

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 表示禁用超时;

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

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

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

client = httpx.AsyncClient(timeout=None)
或者
httpx.get(url=url, timeout=None)

2.2 错误：SSL: CERTIFICATE_VERIFY_FAILED

• 在多域名或者通配符域名证书的情况下，可能会出现此类错误。
• 参考方案： 初始化时将 verify 赋值为 False

client = httpx.AsyncClient(verify=False)
 或者
httpx.get(url=url, verify=False)

2.3 综合

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

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
with httpx.Client() as client:
    ...

# 使用方法2
client = httpx.Client()
try:
    ...
finally:
    client.close()

3.2 发出请求

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

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

import httpx

with httpx.Client(base_url='http://httpbin.org') as client:
     r = client.get('/headers')

print(r.request.url)

3.4 python_web

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 实例：

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

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

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

2.6 进度条

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

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