本文最后更新于：2021年3月2日上午

信息

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

官方文档：https://www.python-httpx.org/
github: https://github.com/encode/httpx/

安装

版本需求：Python 3.6+

1	`pip install httpx`

如果希望支持 HTTP/2，可以使用：

1	`pip install httpx[http2]`

如果希望支持 brotli 压缩算法，可以使用：

1	`$ pip install httpx[brotli]`

快速开始

请求

基础请求方法

基础的请求方法都类似

r = httpx.get('https://httpbin.org/get')
r = httpx.post('https://httpbin.org/post', data={'key': 'value'})
r = httpx.put('https://httpbin.org/put', data={'key': 'value'})
r = httpx.delete('https://httpbin.org/delete')
r = httpx.head('https://httpbin.org/get')
r = httpx.options('https://httpbin.org/get')

url = 'http://httpbin.org/headers'
headers = {'user-agent': 'my-app/0.0.1'}
r = httpx.get(url, headers=headers)
print(r.json())

get请求参数

1
2
3

params = {'key1': 'value1', 'key2': 'value2', 'key3': ['value3', 'value4']}
r = httpx.get('https://httpbin.org/get', params=params)
print(r.url)

post请求数据

1
2
3

data = {'key1': 'value1', 'key2': 'value2', 'key3': ['value3', 'value4']}
r = httpx.post("https://httpbin.org/post", data=data)
print(r.json()['form'])

post请求上传文件

# 文件可以传入文件名和文件类型
files = {'upload-file': ('README.md', open('README.md', 'rb'), 'text/plain')}
r = httpx.post("https://httpbin.org/post", files=files)
print(r.text)

可以一次性传输多个文件

1
2
3

files = [('images', ('foo.png', open('foo.png', 'rb'), 'image/png')),
                      ('images', ('bar.png', open('bar.png', 'rb'), 'image/png'))]
r = httpx.post("https://httpbin.org/post", files=files)

post请求发送json

1
2
3

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

流式响应

二进制

1
2
3

with httpx.stream("GET", "https://www.example.com") as r:
    for data in r.iter_bytes():
        print(data)

文本

1
2
3

with httpx.stream("GET", "https://www.example.com") as r:
    for text in r.iter_text():
        print(text)

多行文本

1
2
3

with httpx.stream("GET", "https://www.example.com") as r:
    for line in r.iter_lines():
        print(line)

响应源码
未被经过 gzip, deflate, brotli 解压缩的源码

1
2
3

with httpx.stream("GET", "https://www.example.com") as r:
    for chunk in r.iter_raw():
        print(chunk)

*注意**
如果您以任何上述方式使用流式响应，则response.content和response.text属性将不可用，并且在访问时会引发错误。
但是，你可以使用响应流功能来有条件地加载响应主体：
1
2
3
4
with httpx.stream("GET", "https://www.example.com") as r: if r.headers['Content-Length'] < 99: r.read() print(r.text)
超时
实际上，如果不进行特别的设置，这个超时的时间时5秒
1
httpx.get('https://github.com/', timeout=0.001)
你可以通过将它设置为None来禁用它，虽然这可能导致永远挂起
1
httpx.get('https://github.com/', timeout=None)
授权验证

明文身份验证

1	`httpx.get("https://example.com", auth=("my_user", "password123"))`

Digest 身份验证

1 2	`auth = httpx.DigestAuth("my_user", "password123") httpx.get("https://example.com", auth=auth)`

响应

内容获取

二进制数据
程序会自动处理 gzip 和 deflate 压缩的响应
如果安装了 brotlipy，那么 brotli 压缩的响应也能够处理
1
2
3
r = httpx.get('https://www.example.org/') print(r.content)

文本

1
2
3

r = httpx.get('https://www.example.org/')
print(r.encoding)
print(r.text)

通常情况下都是utf-8，遇到不是的话，需要手动指定

1	`r.encoding = 'ISO-8859-1'`

json

1 2	`r = httpx.get('https://api.github.com/events') print(r.json())`

状态码

1 2	`r = httpx.get('https://httpbin.org/get') print(r.status_code)`

对于正常的响应，httpx有非常简单的判断方法

1 2	`is_ok = r.status_code == httpx.codes.OK print(is_ok)`

对于异常的响应，也有简易的抛出错误的方法
这个方法若是响应正常，会返回None.若是出现问题，则会抛出对应错误

1	`r.raise_for_status()`

响应头

1 2	`r = httpx.get('https://httpbin.org/get') print(r.headers)`

重定向

# 由http跳转到https
r = httpx.get('http://github.com/')
print(r.url)
print(r.status_code)
print(r.history)

如果不想要跳转，那么可以设置

1
2
3

r = httpx.get('http://github.com/', allow_redirects=False)
print(r.status_code)
print(r.history)

在使用head方式发送请求时，也能用这个参数来启用跳转

1
2
3

r = httpx.head('http://github.com/', allow_redirects=True)
print(r.url)
print(r.history)

cookies

从响应中获取

1 2	`r = httpx.get('http://httpbin.org/cookies/set?chocolate=chip', allow_redirects=False) print(r.cookies['chocolate'])`

请求时设置(简易)

1
2
3

cookies = {"peanut": "butter"}
r = httpx.get('http://httpbin.org/cookies', cookies=cookies)
print(r.json())

请求时设置(标准)

cookies = httpx.Cookies()
cookies.set('cookie_on_domain', 'hello, there!', domain='httpbin.org')
cookies.set('cookie_off_domain', 'nope.', domain='example.org')
r = httpx.get('http://httpbin.org/cookies', cookies=cookies)
print(r.json())

更多用法

Client 实例

如果你曾经用过Requests，那么你可以将httpx.Client()看作是requests.Session()来快速理解

如果你并不是想要写“一次性访问代码”，那么应该用客户端实例
每次在使用 “快速入门” 里面的顶层API来访问网络时都会建立TCP连接。频繁的建立TCP连接会使得程序的效率会大幅度下降

Client 实例拥有 HTTP连接池。当你访问同一个网络，Client 重用已经创建了的TCP连接。这样做能显著地提高订正API的性能

减少了请求之间的延迟（重用连接无需握手）
减少CPU使用率和往返次数
减少网络拥塞

Client也有其它的功能

Cookie保持
在所有传出请求中应用配置
通过HTTP代理发送请求
使用HTTP/2

使用

推荐使用 with语句来使用

1 2	`with httpx.Client() as client: ...`

也可以显式的在不需要使用用用close函数关闭

client = httpx.Client()
try:
    ...
finally:
    client.close()

请求

Clent拥有和高层级API一样的函数，比如说 get(), post()，在 “快速开始” 中的方法大多都有实现

1 2	`with httpx.Client() as client: r = client.get('https://example.com')`

共享请求配置

可以为Client配置内容，其配置的内容在发送请求时也会应用

url = 'http://httpbin.org/headers'
headers = {'user-agent': 'my-app/0.0.1'}
with httpx.Client(headers=headers) as client:
    r = client.get(url)
r.json()['headers']['User-Agent']

此处get()并没有传入headers，但由于Client存在headers配置，最终的GET请求也存在headers配置

配置合并

当你同时在Client中配置内容，在请求函数中也配置内容，可能会发生两种情况：

对于 Headers, Cookies, Query参数
它们的值会合并并一起应用到最后的请求中

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)

print(r.request.url)
print(r.request.headers['X-Auth'])
print(r.request.headers['X-Custom'])

对于其它参数，请求函数中设置的参数会覆盖Client中设置的参数

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
base64.b64decode(auth)
b'alice:ecila123'

基础url

基础url可以让请求输入参数时少输入一些url

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

print(r.request.url)

更多的Client内容可以查看API

Request 实例

httpx 是支持显式创建 Request实例的

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

创建的 Request实例可以使用Client实例的 send()方法来发送

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

如果想要使用不同层级的配置，可以使用Client实例的build_request()方法

headers = {"X-Api-Key": "...", "X-Client-ID": "ABC123"}

with httpx.Client(headers=headers) as client:
    request = client.build_request("GET", "https://example.com")
    print(request.headers["X-Client-ID"])  # "ABC123"
    # 删除header中的一个项
    request.headers.pop("X-Api-Key")
    response = client.send(request)
    ...

事件钩子 Event Hooks

httpx 可以让你注册 “事件钩子”，每当指定的事件发生，钩子内容会被调用
现在能够使用的钩子有两个

request - 请求即将发送时调用。传递请求实例
response - 响应返回后调用。传递响应实例

例：利用事件钩子进行记录

def log_request(request):
    print(f"Request event hook: {request.method} {request.url} - Waiting for response")

def log_response(response):
    request = response.request
    print(f"Response event hook: {request.method} {request.url} - Status {response.status_code}")

client = httpx.Client(event_hooks={'request': [log_request], 'response': [log_response]})

例：在状态码为4xx或5xx时，抛出错误

def raise_on_4xx_5xx(response):
    response.raise_for_status()

client = httpx.Client(event_hooks={'response': [raise_on_4xx_5xx]})

钩子能够传入多个函数，只要一并放入参数中即可

1
2
3

client = httpx.Client()
client.event_hooks['request'] = [log_request]
client.event_hooks['response'] = [log_response, raise_on_4xx_5xx]

在设置完毕后，可以使用Client对象的event_hooks属性查看设置

1
2
3

client = httpx.Client()
client.event_hooks['request'] = [log_request]
client.event_hooks['response'] = [log_response, raise_on_4xx_5xx]

如果未来使用异步请求，那么钩子函数也需要是异步函数

监视响应进度

如果需要监视响应进度（例如下载文件时的下载进度），可以使用 stream请求和response.num_bytes_downloaded

例：使用 tqdm库配合显示下载进度

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

.netrc支持

HTTPX支持 .netrc文件。在trust_env = True情况下，如果未定义auth参数，则HTTPX尝试将auth从.netrc文件添加到请求的标头中
默认情况下 trust_env参数值为True

如果NETRC环境没设置，那么 httpx 会尝试去读取系统默认环境下的文件

1
2
3

# 更改环境
import os
os.environ["NETRC"] = "my_default_folder/.my_netrc"

注意，如果使用Client，那么trust_env参数应该在Client中进行设置，而不是在请求方法中传入

HTTP 代理

httpx支持使用代理，传入proxies传入内容即可使用
例: 传入HTTP代理与 HTTPS代理

proxies = {
    "http://": "http://localhost:8030",
    "https://": "http://localhost:8031",
}

with httpx.Client(proxies=proxies) as client:
    ...

代理路由

httpx提供多种不同的选择代理的方法

通配
让所有请求都走这个代理

1
2
3

proxies = {
    "all://": "http://localhost:8030",
}

协议匹配
根据请求类型走不同的代理

proxies = {
    "http://": "http://localhost:8030",
    "https://": "http://localhost:8031",
}

域名匹配
域名为example.com的请求走这个代理

1
2
3

proxies = {
    "all://example.com": "http://localhost:8030",
}

域名为example.com的HTTP请求走这个代理

1
2
3

proxies = {
    "http://example.com": "http://localhost:8030",
}

域名及子域名匹配example.com的请求走这个代理

1
2
3

proxies = {
    "all://*example.com": "http://localhost:8030",
}

example.com子域名的请求都走这个代理

1
2
3

proxies = {
    "all://*.example.com": "http://localhost:8030",
}

端口匹配
访问”example.com”的1234端口的HTTPS请求走这个代理

1
2
3

proxies = {
    "https://example.com:1234": "http://localhost:8030",
}

所有1234端口的请求走这个代理

1
2
3

proxies = {
    "all://*:1234": "http://localhost:8030",
}

不走代理
只要设定键值为None即可

proxies = {
    # Route requests through a proxy by default...
    "all://": "http://localhost:8031",
    # Except those for "example.com".
    "all://example.com": None,
}

代理类型

代理根据它的流程有两种类型：转发与隧道

转发：就只是单纯的转发
隧道：中间人，即将客户端的数据包解析出来，再将数据重新发送。这样子包的记录信息（IP、Mac等）都时代理的信息

默认情况下 httpx 传输 HTTP时会使用转发类型的代理，而传输HTTPS则使用隧道类型的代理

httpx提供了代理检查功能，通过使用httpx.Proxy实例，设置 FORWARD_ONLY 或 TUNNEL_ONLY

# 所有HTTPS请求都走这个代理, 这个代理必须是隧道类型
proxies = httpx.Proxy(
    url="https://localhost:8030",
    mode="TUNNEL_ONLY",
)

with httpx.Client(proxies=proxies) as client:
    # This HTTP request will be tunneled instead of forwarded.
    r = client.get("http://example.com")

Timeout

实际上，除了直接设置总体超时时间，httpx还可以对超时进行更细致的设定
超时有四种类型：

连接超时 connect
指定等待与请求的主机建立连接之前的最长时间。如果HTTPX在此时间段内无法连接，则会引发ConnectTimeout异常
读取超时 read
指定了等待接收数据（例如，响应主体的块）的最长时间。如果HTTPX在此时间段内无法接收数据，则会引发ReadTimeout异常
写超时 write
指定等待发送数据块（例如，请求正文的块）的最大持续时间。如果HTTPX在此时间段内无法发送数据，则会引发WriteTimeout异常
从连接池获取连接超时 pool
指定等待从连接池获取连接的最大持续时间。如果HTTPX在此时间段内无法获取连接，则会引发PoolTimeout异常

# A client with a 60s timeout for connecting, and a 10s timeout elsewhere.
timeout = httpx.Timeout(10.0, connect=60.0)
client = httpx.Client(timeout=timeout)

response = client.get('http://example.com/')

连接池限制

你可以设置httpx连接池连接数量限制

max_keepalive 允许的保持活动连接数。为None则表示无限制（默认值：10）
max_connections 最大链接数。为None则表示无限制（默认值：100）

1 2	`limits = httpx.Limits(max_keepalive_connections=5, max_connections=10) client = httpx.Client(limits=limits)`

SSL证书

通过HTTPS发出请求时，HTTPX需要验证所请求主机的身份。为此，它使用由受信任的证书颁发机构（CA）交付的SSL证书捆绑包（也称为CA bundle）

默认情况下，httpx 使用 Certifi 的 CA bundle
在大多数情况下，这个设置已经足够了。但在一些情况可能需要使用不同的CA bundle
如果想要使用自己的CA bundle，可以通过 verify 参数传入

1
2
3

import httpx

r = httpx.get("https://example.org", verify="path/to/client.pem")

或者使用标准库ssl.SSLContext

import ssl
import httpx
context = ssl.create_default_context()
context.load_verify_locations(cafile="/tmp/client.pem")
httpx.get('https://example.org', verify=context)

httpx 也拥有方便创建 SSLContext 的函数
create_ssl_context函数接受与 httpx.Client 或httpx.AsyncClient 相同的SSL配置参数集（trust_env，verify，cert和http2参数）

1
2
3

import httpx
context = httpx.create_ssl_context(verify="/tmp/client.pem")
httpx.get('https://example.org', verify=context)

如果你完全不需要SSL，可以通过对 verify 参数传入False来进行设置

1
2
3

import httpx

r = httpx.get("https://example.org", verify=False)

SSL配置与Client实例

如果你使用Client实例，并且希望使用SSL配置，那么应该在创建它的时候传入
请求函数（get,post等）并不支持对SSL配置进行修改。如果在不同情况下需要不同的SSL设置，则应该使用多个客户端实例，对每个实例进行不同的设置。每个客户端将在该池中的所有连接上使用具有特定的固定SSL配置的隔离连接池

HTTPS请求与本地服务器

向本地服务器（例如在本地主机上运行的开发服务器）发出请求时，通常将使用未加密的HTTP连接

如果需要建立与本地服务器的HTTPS连接（例如，测试仅HTTPS服务），则需要创建并使用自己的证书（自签证书）
在服务器生成密钥对以后，拿到其对应的.pem密钥文件，传入verify参数之中即可

1 2	`import httpx r = httpx.get("https://localhost:8000", verify="/tmp/client.pem")`

自定义传输

Client能接受传输参数，该参数允许您提供一个自定义的传输对象，该对象将用于执行请求的实际发送

对于某些高级配置，需要直接实例化传输类，并将其传递给客户端实例。 httpcore 软件包提供了只能通过此低级API使用的local_address配置

import httpx, httpcore
ssl_context = httpx.create_ssl_context()
transport = httpcore.SyncConnectionPool(
    ssl_context=ssl_context,
    max_connections=100,
    max_keepalive_connections=20,
    keepalive_expiry=5.0,
    local_address="0.0.0.0"
)  # 使用httpx默认标准设置, 然后添加了只允许 IPv4  的'local_address'设置
client = httpx.Client(transport=transport)

类似地，httpcore提供了一个uds选项，用于通过Unix域套接字进行连接，该选项仅可通过以下底层API使用：

import httpx, httpcore
ssl_context = httpx.create_ssl_context()
transport = httpcore.SyncConnectionPool(
    ssl_context=ssl_context,
    max_connections=100,
    max_keepalive_connections=20,
    keepalive_expiry=5.0,
    uds="/var/run/docker.sock",
)  # 通过 Unix Socket 连接 Docker API
client = httpx.Client(transport=transport)
response = client.get("http://docker/info")
response.json()

与 Client不同，较低级别的httpcore传输实例不包含用于配置方面（例如连接池详细信息）的任何默认值，因此在使用此API时需要提供更明确的配置

自定义传输类

传输实例必须实现 httpcore 定义的传输API
自定义传输类应该继承httpcore.AsyncHTTPTransport来实现与AsyncClient一起使用的传输
或者继承httpcore.SyncHTTPTransport来实现与Client一起使用的传输

定制传输实现的完整示例为：

import json
import httpcore


class HelloWorldTransport(httpcore.SyncHTTPTransport):
    """
    A mock transport that always returns a JSON "Hello, world!" response.
    """

    def request(self, method, url, headers=None, stream=None, timeout=None):
        message = {"text": "Hello, world!"}
        content = json.dumps(message).encode("utf-8")
        stream = httpcore.PlainByteStream(content)
        headers = [(b"content-type", b"application/json")]
        return b"HTTP/1.1", 200, b"OK", headers, stream

上面代码的作用与下面的代码相同

1
2
3

import httpx
client = httpx.Client(transport=HelloWorldTransport())
response = client.get("https://example.org/")

异步

异步是一种并发模型，其效率远远高于多线程，并且可以提供显着的性能优势并允许使用长期存在的网络连接（例如WebSockets）

想要用 httpx 发出异步请求，则需要使用AsyncClient

1 2	`async with httpx.AsyncClient() as client: r = await client.get('https://www.example.com/')`

有AsyncClient有很多与Client相似的请求方法

AsyncClient.get(url, …)
AsyncClient.options(url, …)
AsyncClient.head(url, …)
AsyncClient.post(url, …)
AsyncClient.put(url, …)
AsyncClient.patch(url, …)
AsyncClient.delete(url, …)
AsyncClient.request(method, url, …)
AsyncClient.send(request, …)

关闭

AsyncClient的关闭的方法时aclose()

1 2	`client = httpx.AsyncClient() await client.aclose()`

但是，实际上依旧推荐使用with来使用，并不推荐显式关闭

流式响应

client = httpx.AsyncClient()
async with client.stream('GET', 'https://www.example.com/') as response:
    async for chunk in response.aiter_bytes():
        ...

Response.aread()
用于有条件地读取流块内的响应
Response.aiter_bytes()
用于以字节形式流式传输响应内容
Response.aiter_text()
用于以文本形式流式传输响应内容
Response.aiter_lines()
用于将响应内容作为文本流传输
Response.aiter_raw()
用于流式传输原始响应字节，而无需应用内容解码
Response.aclose()
用于关闭响应。通常不需要这样做，因为.stream块会在退出时自动关闭响应

对于无法使用上下文块的情况，可以通过使用client.send(...，stream = True)发送一个 Request实例来进入“手动模式”

import httpx
from starlette.background import BackgroundTask
from starlette.responses import StreamingResponse

client = httpx.AsyncClient()

async def home(request):
    req = client.build_request("GET", "https://www.example.com/")
    r = await client.send(req, stream=True)
    return StreamingResponse(r.aiter_text(), background=BackgroundTask(r.aclose))

AsyncIO

import asyncio
import httpx

async def main():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://www.example.com/')
        print(response)

asyncio.run(main())

HTTP2

HTTP/2 提供了更有效的传输，并具有潜在的性能优势。HTTP2不会更改请求或响应的核心语义，但会更改数据与服务器之间的发送方式

详细可以参考：我的HTTP2记录、HTTP2解析

httpx 默认并不支持 HTTP/2，如果想要使用，则需要启用HTTP/2支持
首先，需要安装 httpx 的 HTTP/2依赖

1	`pip install httpx[http2]`

再带那么重初始化支持 HTTP/2 的 Client

1 2	`async with httpx.AsyncClient(http2=True) as client: ...`

Client 和 AsyncClient 都可以使用 HTTP/2 支持

检查HTTP版本

在Client上启用 HTTP/2 支持不一定意味着请求和响应将通过 HTTP/2 传输，因为只有客户端和服务器都支持 HTTP/2，HTTP2才会运行。如果连接到仅支持HTTP / 1.1的服务器，那么Client会自动改用 HTTP/1.1 进行连接

通过检查Response.http_version属性来确定使用了哪个版本的HTTP协议

1
2
3

client = httpx.AsyncClient(http2=True)
response = await client.get(...)
print(response.http_version)  # "HTTP/1.0", "HTTP/1.1", or "HTTP/2".

后续可继续查阅官方文档内容

python httpx

本博客所有文章除特别声明外，均采用 CC BY-SA 4.0 协议，转载请注明出处！

Javascript-AAencode-混淆-分析上一篇

HTTP2 下一篇

Python-httpx

信息

安装

快速开始

请求

基础请求方法

自定义header

get请求参数

post请求数据

post请求上传文件

post请求发送json

流式响应

超时

授权验证

响应

内容获取

状态码

响应头

重定向

cookies

更多用法

Client 实例

使用

请求

共享请求配置

配置合并

基础url

Request 实例

事件钩子 Event Hooks

监视响应进度

.netrc支持

HTTP 代理

代理路由

代理类型

Timeout

连接池限制

SSL证书

SSL配置与Client实例

HTTPS请求与本地服务器

自定义传输

自定义传输类

异步

关闭

流式响应

AsyncIO

HTTP2

检查HTTP版本