Skip to content

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.ConnectErrorhttpx.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:
    ...

有关 clientroot_path 参数的更多详情,请参阅 ASGI 文档

ASGI 启动与关闭

HTTPX 的职责范围不包含触发应用程序的 ASGI 生命周期事件。

建议配合使用 asgi-lifespanLifespanManagerAsyncClient

自定义传输器

传输器实例必须实现底层的 Transport API,该 API 负责发送单个请求并返回响应。您应该继承 httpx.BaseTransport 来实现与 Client 配合使用的传输器,或者继承 httpx.AsyncBaseTransport 来实现与 AsyncClient 配合使用的异步传输器。

在传输器 API 这一层,我们使用熟悉的 RequestResponse 模型。

有关 Transport API 具体细节的更多信息,请参阅 handle_requesthandle_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)

对于更高级的用例,您可以参考 第三方模拟库 RESPXpytest-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_PROXYNO_PROXY的文档。