HTTPX 的 Client
也接受 transport
参数。该参数允许你提供一个自定义的 Transport 对象,用于实际执行请求的发送操作。
HTTP 传输层
对于某些高级配置,你可能需要直接实例化一个传输类,并将其传递给客户端实例。例如 local_address
配置就只通过这种底层 API 提供。
>>> import httpx
>>> transport = httpx.HTTPTransport(local_address="0.0.0.0")
>>> client = httpx.Client(transport=transport)
连接重试机制也通过此接口提供。当发生 httpx.ConnectError
或 httpx.ConnectTimeout
时,请求会按指定次数重试,从而在不稳定的网络环境下实现更流畅的操作。如果需要其他类型的重试行为(如处理读写错误或响应 503 Service Unavailable
),可以考虑使用通用工具如 tenacity。
>>> import httpx
>>> transport = httpx.HTTPTransport(retries=1)
>>> client = httpx.Client(transport=transport)
同样地,直接实例化传输层还提供了 uds
选项,用于通过 Unix 域套接字连接,该功能也仅通过此底层 API 提供:
>>> import httpx
>>> # 通过 Unix 套接字连接 Docker API
>>> transport = httpx.HTTPTransport(uds="/var/run/docker.sock")
>>> client = httpx.Client(transport=transport)
>>> response = client.get("http://docker/info")
>>> response.json()
{"ID": "...", "Containers": 4, "Images": 74, ...}
WSGI 传输层
你可以配置 httpx
客户端,使其直接通过 WSGI 协议调用 Python web 应用程序。
这在以下两种主要场景中特别有用:
- 在测试用例中使用
httpx
作为客户端 - 在测试、开发或预发布环境中模拟外部服务
示例
以下是一个与 Flask 应用集成的示例:
from flask import Flask
import httpx
app = Flask(__name__)
@app.route("/")
def hello():
return "Hello World!"
transport = httpx.WSGITransport(app=app)
with httpx.Client(transport=transport, base_url="http://testserver") as client:
r = client.get("/")
assert r.status_code == 200
assert r.text == "Hello World!"
配置
对于更复杂的场景,您可能需要自定义 WSGI 传输器。这允许您:
- 通过设置
raise_app_exceptions=False
来检查 500 错误响应,而不是抛出异常 - 通过设置
script_name
(WSGI) 将 WSGI 应用挂载到子路径 - 通过设置
remote_addr
(WSGI) 为请求指定客户端地址
例如:
# 实例化一个客户端,该客户端使用 "1.2.3.4" 作为客户端 IP 发起 WSGI 请求
transport = httpx.WSGITransport(app=app, remote_addr="1.2.3.4")
with httpx.Client(transport=transport, base_url="http://testserver") as client:
...
ASGI 传输器
您可以将 httpx
客户端配置为使用 ASGI 协议直接调用异步 Python Web 应用。
这在以下两种主要场景中特别有用:
- 在测试用例中使用
httpx
作为客户端 - 在测试、开发或预发布环境中模拟外部服务
示例
我们以这个 Starlette 应用为例:
from starlette.applications import Starlette
from starlette.responses import HTMLResponse
from starlette.routing import Route
async def hello(request):
return HTMLResponse("Hello World!")
app = Starlette(routes=[Route("/", hello)])
我们可以直接对该应用发起请求,如下所示:
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
r = await client.get("/")
assert r.status_code == 200
assert r.text == "Hello World!"
配置
对于一些更复杂的场景,您可能需要自定义 ASGI 传输层。这允许您:
- 通过设置
raise_app_exceptions=False
来检查 500 错误响应而不是抛出异常 - 通过设置
root_path
将 ASGI 应用挂载到子路径 - 通过设置
client
为请求指定客户端地址
例如:
# 实例化一个客户端,该客户端使用 IP "1.2.3.4" 和端口 123 发起 ASGI 请求
transport = httpx.ASGITransport(app=app, client=("1.2.3.4", 123))
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
...
有关 client
和 root_path
参数的更多详情,请参阅 ASGI 文档。
ASGI 启动与关闭
HTTPX 的职责范围不包含触发应用程序的 ASGI 生命周期事件。
建议配合使用 asgi-lifespan 的 LifespanManager
和 AsyncClient
。
自定义传输器
传输器实例必须实现底层的 Transport API,该 API 负责发送单个请求并返回响应。您应该继承 httpx.BaseTransport
来实现与 Client
配合使用的传输器,或者继承 httpx.AsyncBaseTransport
来实现与 AsyncClient
配合使用的异步传输器。
在传输器 API 这一层,我们使用熟悉的 Request
和 Response
模型。
有关 Transport API 具体细节的更多信息,请参阅 handle_request
和 handle_async_request
的文档字符串。
以下是一个完整自定义传输器实现的示例:
import json
import httpx
class HelloWorldTransport(httpx.BaseTransport):
"""
一个模拟传输器,始终返回 JSON 格式的 "Hello, world!" 响应。
"""
def handle_request(self, request):
return httpx.Response(200, json={"text": "Hello, world!"})
或者这个示例,它使用自定义传输器和 httpx.Mounts
来始终重定向 http://
请求:
class HTTPSRedirect(httpx.BaseTransport):
"""
一个始终重定向到 HTTPS 的传输器。
"""
def handle_request(self, request):
url = request.url.copy_with(scheme="https")
return httpx.Response(303, headers={"Location": str(url)})
将所有 http
请求重定向到 https
的客户端
transport = httpx.Mounts({ 'http://': HTTPSRedirect() 'https://': httpx.HTTPTransport() }) client = httpx.Client(transport=transport)
这里有一个有用的模式是包装默认 HTTP 实现的定制传输类。例如...
```python
class DebuggingTransport(httpx.BaseTransport):
def __init__(self, **kwargs):
self._wrapper = httpx.HTTPTransport(**kwargs)
def handle_request(self, request):
print(f">>> {request}")
response = self._wrapper.handle_request(request)
print(f"<<< {response}")
return response
def close(self):
self._wrapper.close()
transport = DebuggingTransport()
client = httpx.Client(transport=transport)
这是另一个案例,我们在多个不同代理之间使用轮询机制...
class ProxyRoundRobin(httpx.BaseTransport):
def __init__(self, proxies, **kwargs):
self._transports = [
httpx.HTTPTransport(proxy=proxy, **kwargs)
for proxy in proxies
]
self._idx = 0
def handle_request(self, request):
transport = self._transports[self._idx]
self._idx = (self._idx + 1) % len(self._transports)
return transport.handle_request(request)
def close(self):
for transport in self._transports:
transport.close()
proxies = [
httpx.Proxy("http://127.0.0.1:8081"),
httpx.Proxy("http://127.0.0.1:8082"),
httpx.Proxy("http://127.0.0.1:8083"),
]
transport = ProxyRoundRobin(proxies=proxies)
client = httpx.Client(transport=transport)
模拟传输器
在测试过程中,能够模拟传输器并返回预设响应(而非实际发起网络请求)通常非常有用。
httpx.MockTransport
类接受一个处理器函数,可用于将请求映射到预设响应:
def handler(request):
return httpx.Response(200, json={"text": "Hello, world!"})
# 如果设置了 TESTING 环境变量,则切换到模拟传输器
if os.environ.get('TESTING', '').upper() == "TRUE":
transport = httpx.MockTransport(handler)
else:
transport = httpx.HTTPTransport()
client = httpx.Client(transport=transport)
对于更高级的用例,您可以参考 第三方模拟库 RESPX 或 pytest-httpx 库。
挂载传输器
您还可以针对特定协议或域名挂载传输器,通过 与代理路由相同的配置方式 来控制出站请求应路由到哪个传输器。
import httpx
class HTTPSRedirectTransport(httpx.BaseTransport):
"""
始终重定向到 HTTPS 的传输器
"""
def handle_request(self, method, url, headers, stream, extensions):
scheme, host, port, path = url
if port is None:
location = b"https://%s%s" % (host, path)
else:
location = b"https://%s:%d%s" % (host, port, path)
stream = httpx.ByteStream(b"")
headers = [(b"location", location)]
extensions = {}
return 303, headers, stream, extensions
将所有 http
请求重定向到 https
的客户端
mounts = {'http://': HTTPSRedirectTransport()} client = httpx.Client(mounts=mounts)
以下是几个利用挂载传输器的其他示例场景...
在单个指定域名上禁用 HTTP/2...
```python
mounts = {
"all://": httpx.HTTPTransport(http2=True),
"all://*example.org": httpx.HTTPTransport()
}
client = httpx.Client(mounts=mounts)
模拟对指定域名的请求:
# 所有对 "example.org" 的请求将被模拟处理
# 其他请求正常执行
def handler(request):
return httpx.Response(200, json={"text": "Hello, World!"})
mounts = {"all://example.org": httpx.MockTransport(handler)}
client = httpx.Client(mounts=mounts)
添加对自定义协议的支持:
# 支持类似 "file:///Users/sylvia_green/websites/new_client/index.html" 的 URL
mounts = {"file://": FileSystemTransport()}
client = httpx.Client(mounts=mounts)
路由机制
HTTPX 提供了强大的请求路由机制,允许您编写复杂规则来指定每个请求应使用的传输器。
mounts
字典将 URL 模式映射到 HTTP 传输器。HTTPX 会根据 URL 模式匹配请求的 URL,决定使用哪个传输器(如果有)。匹配过程从最具体的 URL 模式(如 https://<域名>:<端口>
)到最不具体的模式(如 https://
)依次进行。
HTTPX 支持基于协议、域名、端口或这些组合来路由请求。
通配符路由
将所有请求通过特定传输器路由...
mounts = {
"all://": httpx.HTTPTransport(proxy="http://localhost:8030"),
}
协议路由
将 HTTP 请求通过一个传输器路由,HTTPS 请求通过另一个...
mounts = {
"http://": httpx.HTTPTransport(proxy="http://localhost:8030"),
"https://": httpx.HTTPTransport(proxy="http://localhost:8031"),
}
域名路由
代理所有发往 "example.com" 域名的请求,其他请求直接通过...
mounts = {
"all://example.com": httpx.HTTPTransport(proxy="http://localhost:8030"),
}
仅代理 "example.com" 域名的 HTTP 请求,HTTPS 和其他请求直接通过...
mounts = {
"http://example.com": httpx.HTTPTransport(proxy="http://localhost:8030"),
}
代理所有发往 "example.com" 及其子域名的请求,其他请求直接通过...
mounts = {
"all://*example.com": httpx.HTTPTransport(proxy="http://localhost:8030"),
}
仅代理严格子域名(如 "sub.example.com")的请求,"example.com" 和其他请求直接通过...
mounts = {
"all://*.example.com": httpx.HTTPTransport(proxy="http://localhost:8030"),
}
端口路由
代理发往 "example.com" 端口 1234 的 HTTPS 请求...
mounts = {
"https://example.com:1234": httpx.HTTPTransport(proxy="http://localhost:8030"),
}
代理所有发往端口 1234 的请求...
mounts = {
"all://*:1234": httpx.HTTPTransport(proxy="http://localhost:8030"),
}
无代理支持
也可以定义哪些请求_不应该_通过代理传输。
为此,将代理 URL 设为 None
。例如...
mounts = {
# 默认所有请求通过代理...
"all://": httpx.HTTPTransport(proxy="http://localhost:8031"),
# 但 "example.com" 除外
"all://example.com": None,
}
复杂配置示例
您可以组合上述路由功能来构建复杂的代理路由配置。例如...
mounts = {
# 默认情况下所有流量都通过代理路由...
"all://": httpx.HTTPTransport(proxy="http://localhost:8030"),
# 但对"domain.io"的HTTPS请求不使用代理...
"https://domain.io": None,
# 对"example.com"及其子域使用另一个代理...
"all://*example.com": httpx.HTTPTransport(proxy="http://localhost:8031"),
# 如果使用HTTP协议,
# 并且请求的是端口5550上的"internal"子域,则使用另一个代理...
"http://internal.example.com:5550": httpx.HTTPTransport(proxy="http://localhost:8032"),
}
环境变量
还可以使用环境变量来控制客户端的mounts字典。这些变量可用于配置客户端的HTTP代理。
更多信息请参阅关于HTTP_PROXY
, HTTPS_PROXY
, ALL_PROXY
和NO_PROXY
的文档。