Skip to content

交通

¥Transports

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

¥HTTPX's Client also accepts a transport argument. This argument allows you to provide a custom Transport object that will be used to perform the actual sending of the requests.

HTTP 传输

¥HTTP Transport

对于某些高级配置，你可能需要直接实例化一个传输类，并将其传递给客户端实例。例如local_address仅可通过此低级 API 获得的配置。

¥For some advanced configuration you might need to instantiate a transport class directly, and pass it to the client instance. One example is the local_address configuration which is only available via this low-level API.

>>> import httpx
>>> transport = httpx.HTTPTransport(local_address="0.0.0.0")
>>> client = httpx.Client(transport=transport)

连接重试也可通过此接口实现。如果出现以下情况，请求将重试指定的次数：httpx.ConnectError或httpx.ConnectTimeout允许在不稳定的网络环境下更顺畅地运行。如果您需要其他形式的重试行为，例如处理读/写错误或响应503 Service Unavailable考虑通用工具，例如韧性。

¥Connection retries are also available via this interface. Requests will be retried the given number of times in case an httpx.ConnectError or an httpx.ConnectTimeout occurs, allowing smoother operation under flaky networks. If you need other forms of retry behaviors, such as handling read/write errors or reacting to 503 Service Unavailable, consider general-purpose tools such as tenacity.

>>> import httpx
>>> transport = httpx.HTTPTransport(retries=1)
>>> client = httpx.Client(transport=transport)

类似地，直接实例化传输提供了uds通过 Unix 域套接字进行连接的选项仅可通过此低级 API 获得：

¥Similarly, instantiating a transport directly provides a uds option for connecting via a Unix Domain Socket that is only available via this low-level API:

>>> import httpx
>>> # Connect to the Docker API via a Unix Socket.
>>> 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 传输

¥WSGI Transport

您可以配置httpx客户端使用 WSGI 协议直接调用 Python Web 应用程序。

¥You can configure an httpx client to call directly into a Python web application using the WSGI protocol.

这对于两个主要用例特别有用：

¥This is particularly useful for two main use-cases:

  • 使用httpx作为测试用例中的客户端。

    ¥Using httpx as a client inside test cases.

  • 在测试期间、开发或登台环境中模拟外部服务。

    ¥Mocking out external services during tests or in dev or staging environments.

例子

¥Example

以下是针对 Flask 应用程序集成的示例：

¥Here's an example of integrating against a Flask application:

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!"

配置

¥Configuration

对于一些更复杂的情况，你可能需要自定义 WSGI 传输。这允许你：

¥For some more complex cases you might need to customize the WSGI transport. This allows you to:

  • 检查 500 错误响应，而不是通过设置引发异常raise_app_exceptions=False。

    ¥Inspect 500 error responses rather than raise exceptions by setting raise_app_exceptions=False.

  • 通过设置将 WSGI 应用程序挂载到子路径script_name（WSGI）。

    ¥Mount the WSGI application at a subpath by setting script_name (WSGI).

  • 通过设置使用给定的客户端地址进行请求remote_addr（WSGI）。

    ¥Use a given client address for requests by setting remote_addr (WSGI).

例如：

¥For example:

# Instantiate a client that makes WSGI requests with a client IP of "1.2.3.4".
transport = httpx.WSGITransport(app=app, remote_addr="1.2.3.4")
with httpx.Client(transport=transport, base_url="http://testserver") as client:
    ...

ASGI运输

¥ASGI Transport

您可以配置httpx客户端使用 ASGI 协议直接调用异步 Python Web 应用程序。

¥You can configure an httpx client to call directly into an async Python web application using the ASGI protocol.

这对于两个主要用例特别有用：

¥This is particularly useful for two main use-cases:

  • 使用httpx作为测试用例中的客户端。

    ¥Using httpx as a client inside test cases.

  • 在测试期间、开发或登台环境中模拟外部服务。

    ¥Mocking out external services during tests or in dev or staging environments.

例子

¥Example

我们以这个 Starlette 应用程序为例：

¥Let's take this Starlette application as an example:

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)])

我们可以直接向应用程序发出请求，如下所示：

¥We can make requests directly against the application, like so:

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!"

配置

¥Configuration

对于一些更复杂的情况，你可能需要自定义 ASGI 传输。这允许你：

¥For some more complex cases you might need to customise the ASGI transport. This allows you to:

  • 检查 500 错误响应，而不是通过设置引发异常raise_app_exceptions=False。

    ¥Inspect 500 error responses rather than raise exceptions by setting raise_app_exceptions=False.

  • 通过设置将 ASGI 应用程序挂载到子路径root_path。

    ¥Mount the ASGI application at a subpath by setting root_path.

  • 通过设置使用给定的客户端地址进行请求client。

    ¥Use a given client address for requests by setting client.

例如：

¥For example:

# Instantiate a client that makes ASGI requests with a client IP of "1.2.3.4",
# on port 123.
transport = httpx.ASGITransport(app=app, client=("1.2.3.4", 123))
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
    ...

看ASGI 文档欲了解更多详情client和root_path键。

¥See the ASGI documentation for more details on the client and root_path keys.

ASGI启动和关闭

¥ASGI startup and shutdown

触发应用程序的 ASGI 生命周期事件不在 HTTPX 的范围内。

¥It is not in the scope of HTTPX to trigger ASGI lifespan events of your app.

但是建议使用LifespanManager从asgi-lifespan与...成对AsyncClient。

¥However it is suggested to use LifespanManager from asgi-lifespan in pair with AsyncClient.

定制运输

¥Custom transports

传输实例必须实现低级传输 API，该 API 负责发送单个请求并返回响应。您可以将其子类化httpx.BaseTransport实现与之一起使用的传输Client或子类httpx.AsyncBaseTransport实现与之一起使用的传输AsyncClient。

¥A transport instance must implement the low-level Transport API which deals with sending a single request, and returning a response. You should either subclass httpx.BaseTransport to implement a transport to use with Client, or subclass httpx.AsyncBaseTransport to implement a transport to use with AsyncClient.

在传输 API 层，我们使用熟悉的Request和Response模型。

¥At the layer of the transport API we're using the familiar Request and Response models.

查看handle_request和handle_async_request文档字符串以获取有关 Transport API 细节的更多详细信息。

¥See the handle_request and handle_async_request docstrings for more details on the specifics of the Transport API.

自定义传输实现的完整示例如下：

¥A complete example of a custom transport implementation would be:

import json
import httpx

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

    def handle_request(self, request):
        return httpx.Response(200, json={"text": "Hello, world!"})

或者这个例子，它使用自定义传输和httpx.Mounts总是重定向http://要求。

¥Or this example, which uses a custom transport and httpx.Mounts to always redirect http:// requests.

class HTTPSRedirect(httpx.BaseTransport):
    """
    A transport that always redirects to HTTPS.
    """
    def handle_request(self, request):
        url = request.url.copy_with(scheme="https")
        return httpx.Response(303, headers={"Location": str(url)})

# A client where any `http` requests are always redirected to `https`
transport = httpx.Mounts({
    'http://': HTTPSRedirect()
    'https://': httpx.HTTPTransport()
})
client = httpx.Client(transport=transport)

这里一个有用的模式是自定义传输类，它包装了默认的 HTTP 实现。例如……

¥A useful pattern here is custom transport classes that wrap the default HTTP implementation. For example...

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)

这是另一种情况，我们在多个不同的代理上使用循环...

¥Here's another case, where we're using a round-robin across a number of different proxies...

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)

模拟传输

¥Mock transports

在测试期间，模拟传输并返回预定的响应（而不是发出实际的网络请求）通常很有用。

¥During testing it can often be useful to be able to mock out a transport, and return pre-determined responses, rather than making actual network requests.

这httpx.MockTransport类接受一个处理函数，该函数可用于将请求映射到预定的响应上：

¥The httpx.MockTransport class accepts a handler function, which can be used to map requests onto pre-determined responses:

def handler(request):
    return httpx.Response(200, json={"text": "Hello, world!"})


# Switch to a mock transport, if the TESTING environment variable is set.
if os.environ.get('TESTING', '').upper() == "TRUE":
    transport = httpx.MockTransport(handler)
else:
    transport = httpx.HTTPTransport()

client = httpx.Client(transport=transport)

对于更高级的用例，你可能需要查看第三方模拟库 RESPX或pytest-httpx 库。

¥For more advanced use-cases you might want to take a look at either the third-party mocking library, RESPX, or the pytest-httpx library.

安装运输

¥Mounting transports

您还可以针对给定的方案或域安装传输，以控制应通过哪个传输路由传出的请求，使用与指定代理路由相同的样式。

¥You can also mount transports against given schemes or domains, to control which transport an outgoing request should be routed via, with the same style used for specifying proxy routing.

import httpx

class HTTPSRedirectTransport(httpx.BaseTransport):
    """
    A transport that always redirects to 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


# A client where any `http` requests are always redirected to `https`
mounts = {'http://': HTTPSRedirectTransport()}
client = httpx.Client(mounts=mounts)

还有一些关于如何利用安装的运输工具的草图……

¥A couple of other sketches of how you might take advantage of mounted transports...

在单个给定域上禁用 HTTP/2...

¥Disabling HTTP/2 on a single given domain...

mounts = {
    "all://": httpx.HTTPTransport(http2=True),
    "all://*example.org": httpx.HTTPTransport()
}
client = httpx.Client(mounts=mounts)

模拟对给定域的请求：

¥Mocking requests to a given domain:

# All requests to "example.org" should be mocked out.
# Other requests occur as usual.
def handler(request):
    return httpx.Response(200, json={"text": "Hello, World!"})

mounts = {"all://example.org": httpx.MockTransport(handler)}
client = httpx.Client(mounts=mounts)

添加对自定义方案的支持：

¥Adding support for custom schemes:

# Support URLs like "file:///Users/sylvia_green/websites/new_client/index.html"
mounts = {"file://": FileSystemTransport()}
client = httpx.Client(mounts=mounts)

路由

¥Routing

HTTPX 为路由请求提供了强大的机制，允许您编写复杂的规则来指定每个请求应使用哪种传输。

¥HTTPX provides a powerful mechanism for routing requests, allowing you to write complex rules that specify which transport should be used for each request.

这mounts字典将 URL 模式映射到 HTTP 传输协议。HTTPX 将请求的 URL 与 URL 模式进行匹配，以决定应使用哪种传输协议（如果有）。匹配是通过最具体的 URL 模式进行的（例如https://<domain>:<port>）到最不具体的（例如https://）。

¥The mounts dictionary maps URL patterns to HTTP transports. HTTPX matches requested URLs against URL patterns to decide which transport should be used, if any. Matching is done from most specific URL patterns (e.g. https://<domain>:<port>) to least specific ones (e.g. https://).

HTTPX 支持基于以下方式的路由请求方案，领域，港口，或者这些的组合。

¥HTTPX supports routing requests based on scheme, domain, port, or a combination of these.

通配符路由

¥Wildcard routing

通过运输安排一切事宜...

¥Route everything through a transport...

mounts = {
    "all://": httpx.HTTPTransport(proxy="http://localhost:8030"),
}

方案路由

¥Scheme routing

通过一种传输方式路由 HTTP 请求，通过另一种传输方式路由 HTTPS 请求...

¥Route HTTP requests through one transport, and HTTPS requests through another...

mounts = {
    "http://": httpx.HTTPTransport(proxy="http://localhost:8030"),
    "https://": httpx.HTTPTransport(proxy="http://localhost:8031"),
}

域路由

¥Domain routing

代理域“example.com”上的所有请求，让其他请求通过...

¥Proxy all requests on domain "example.com", let other requests pass through...

mounts = {
    "all://example.com": httpx.HTTPTransport(proxy="http://localhost:8030"),
}

代理域“example.com”上的 HTTP 请求，让 HTTPS 和其他请求通过...

¥Proxy HTTP requests on domain "example.com", let HTTPS and other requests pass through...

mounts = {
    "http://example.com": httpx.HTTPTransport(proxy="http://localhost:8030"),
}

将所有请求代理到“example.com”及其子域，让其他请求通过...

¥Proxy all requests to "example.com" and its subdomains, let other requests pass through...

mounts = {
    "all://*example.com": httpx.HTTPTransport(proxy="http://localhost:8030"),
}

将所有请求代理到“example.com”的严格子域，让“example.com”和其他请求通过......

¥Proxy all requests to strict subdomains of "example.com", let "example.com" and other requests pass through...

mounts = {
    "all://*.example.com": httpx.HTTPTransport(proxy="http://localhost:8030"),
}

端口路由

¥Port routing

将端口 1234 上的 HTTPS 请求代理到“example.com”...

¥Proxy HTTPS requests on port 1234 to "example.com"...

mounts = {
    "https://example.com:1234": httpx.HTTPTransport(proxy="http://localhost:8030"),
}

代理端口 1234 上的所有请求...

¥Proxy all requests on port 1234...

mounts = {
    "all://*:1234": httpx.HTTPTransport(proxy="http://localhost:8030"),
}

无代理支持

¥No-proxy support

还可以定义以下请求不应该通过运输进行路由。

¥It is also possible to define requests that shouldn't be routed through the transport.

为此，请通过None作为代理 URL。例如...

¥To do so, pass None as the proxy URL. For example...

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

复杂配置示例

¥Complex configuration example

您可以组合上述路由功能来构建复杂的代理路由配置。例如……

¥You can combine the routing features outlined above to build complex proxy routing configurations. For example...

mounts = {
    # Route all traffic through a proxy by default...
    "all://": httpx.HTTPTransport(proxy="http://localhost:8030"),
    # But don't use proxies for HTTPS requests to "domain.io"...
    "https://domain.io": None,
    # And use another proxy for requests to "example.com" and its subdomains...
    "all://*example.com": httpx.HTTPTransport(proxy="http://localhost:8031"),
    # And yet another proxy if HTTP is used,
    # and the "internal" subdomain on port 5550 is requested...
    "http://internal.example.com:5550": httpx.HTTPTransport(proxy="http://localhost:8032"),
}

环境变量

¥Environment variables

还有一些环境变量可用于控制客户端挂载的字典。它们可用于配置客户端的 HTTP 代理。

¥There are also environment variables that can be used to control the dictionary of the client mounts. They can be used to configure HTTP proxying for clients.

请参阅文档HTTP_PROXY，HTTPS_PROXY，ALL_PROXY和NO_PROXY了解更多信息。

¥See documentation on HTTP_PROXY, HTTPS_PROXY, ALL_PROXY and NO_PROXY for more information.