httpx 自定義傳輸

HTTPX ?Client ?也接受一個(gè)?transport?參數(shù)。此參數(shù)允許您提供將用于執(zhí)行請(qǐng)求實(shí)際發(fā)送的自定義?transport?對(duì)象。

用法

對(duì)于某些高級(jí)配置，您可能需要直接實(shí)例化?transport?類，并將其傳遞到?client?實(shí)例。一個(gè)例子是?local_address?的配置只能通過此低級(jí) API 獲得。

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

也可以通過此接口重試連接。

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

類似地，直接實(shí)例化?transport?提供了一個(gè)?uds?選項(xiàng)，用于通過Unix域套接字進(jìn)行連接，該套接字僅可通過此低級(jí)API進(jìn)行連接：

>>> 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, ...}

urllib3 transport

這個(gè)公共gist提供了一個(gè)使用優(yōu)秀的urllib3庫的?transport?，并且可以與同步的?Client?一起使用...

>>> import httpx
>>> from urllib3_transport import URLLib3Transport
>>> client = httpx.Client(transport=URLLib3Transport())
>>> client.get("https://example.org")
<Response [200 OK]>

編寫自定義transport

?transport?實(shí)例必須實(shí)現(xiàn)低級(jí) ?transport ?API，該API處理發(fā)送單個(gè)請(qǐng)求和返回響應(yīng)。您應(yīng)該創(chuàng)建?httpx?的子類。您應(yīng)該使用子類?httpx.BaseTransport?來實(shí)現(xiàn)與?Client?一起使用的?transport?，或者使用子類?httpx.AsyncBaseTransport?來實(shí)現(xiàn)與?AsyncClient?一起使用的?transport?。

在?transport API ?層，我們使用的是熟悉的?Request?和?Response?模型。

有關(guān)?transport API ?細(xì)節(jié)的更多詳細(xì)信息，請(qǐng)參閱 ?handle_request?和 ?handle_async_request?文檔字符串。

自定義?transport?實(shí)現(xiàn)的完整示例如下：

import json
import httpx


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

    def handle_request(self, request):
        message = {"text": "Hello, world!"}
        content = json.dumps(message).encode("utf-8")
        stream = httpx.ByteStream(content)
        headers = [(b"content-type", b"application/json")]
        return httpx.Response(200, headers=headers, stream=stream)

我們可以以相同的方式使用它：

>>> import httpx
>>> client = httpx.Client(transport=HelloWorldTransport())
>>> response = client.get("https://example.org/")
>>> response.json()
{"text": "Hello, world!"}

模擬transport

在測(cè)試期間，能夠模擬?transport?并返回預(yù)先確定的響應(yīng)，而不是發(fā)出實(shí)際的網(wǎng)絡(luò)請(qǐng)求，這通常很有用。

?httpx.MockTransport?類接受一個(gè)處理程序函數(shù)，該函數(shù)可用于將請(qǐng)求映射到預(yù)先確定的響應(yīng)上：

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)

對(duì)于更高級(jí)的用例，您可能需要查看第三方模擬庫，例如RESPX或pytest-httpx庫。

安裝transport

您還可以針對(duì)給定的方案或域掛載transport，以控制傳出請(qǐng)求應(yīng)通過哪個(gè)transport進(jìn)行路由，其樣式與指定代理路由的樣式相同。

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)

關(guān)于如何利用已安裝的transport工具的其他一些草圖...

正在禁用單個(gè)給定域上的 HTTP/2...

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

模擬對(duì)給定域的請(qǐng)求：

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

添加對(duì)自定義方案的支持：

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