httpx 異步支持

默認(rèn)情況下，HTTPX 提供標(biāo)準(zhǔn)的同步 API，但如果需要，還可以選擇異步?client?。

Async 是一種并發(fā)模型，它比多線程更有效，可以提供顯著的性能優(yōu)勢，并允許使用長期存在的網(wǎng)絡(luò)連接（如 WebSockets）。

如果您使用的是異步 Web 框架，那么您還需要使用異步?client?來發(fā)送傳出 HTTP 請求。

發(fā)出異步請求

要發(fā)出異步請求，您需要一個(gè) ?AsyncClient?。

>>> async with httpx.AsyncClient() as client:
...     r = await client.get('https://www.example.com/')
...
>>> r
<Response [200 OK]>

提示
使用 IPython 或 Python 3.8+使用?python -m ?以交互方式嘗試此代碼，因?yàn)樗鼈冎С?nbsp;?asyncio?在控制臺中執(zhí)行?async ?/?await?表達(dá)式。

接口差異

如果您使用的是異步?client?，那么有一些 API 是使用異步方法的。

請求構(gòu)造

請求方法都是異步的，因此您應(yīng)該對以下所有內(nèi)容使用?response = await client.get(...)?樣式：

• ?AsyncClient.get(url, ...)?
• ?AsyncClient.get(url, ...)?
• ?AsyncClient.get(url, ...)?
• ?AsyncClient.get(url, ...)?
• ?AsyncClient.get(url, ...)?
• ?AsyncClient.get(url, ...)?
• ?AsyncClient.get(url, ...)?
• ?AsyncClient.get(url, ...)?
• ?AsyncClient.get(url, ...)?

打開和關(guān)閉?client?

如果需要上下文管理的client，請使用?async with httpx.AsyncClient()?...

async with httpx.AsyncClient() as client:
    ...

或者，如果要顯式關(guān)閉?client?，請使用?await client.aclose()?:

client = httpx.AsyncClient()
...
await client.aclose()

流式處理響應(yīng)

?AsyncClient.stream(method, url, ...)?方法是一個(gè)異步上下文塊。

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

異步響應(yīng)流式處理方法包括：

• ?Response.aread()?- 用于有條件地讀取流塊內(nèi)的響應(yīng)。
• ?Response.aiter_bytes()?- 用于將響應(yīng)內(nèi)容流式傳輸為字節(jié)。
• ?Response.aiter_text()?- 用于將響應(yīng)內(nèi)容流式傳輸為文本。
• ?Response.aiter_lines()?- 用于將響應(yīng)內(nèi)容流式傳輸為文本行。
• ?Response.aiter_raw()?- 用于流式傳輸原始響應(yīng)字節(jié)，而不應(yīng)用內(nèi)容解碼。
• ?Response.aclose()?- 用于關(guān)閉響應(yīng)。您通常不需要這樣做，因?yàn)?.stream?塊會在退出時(shí)自動關(guān)閉響應(yīng)。

對于上下文塊使用不切實(shí)際的情況，可以通過使用?client.send(..., stream=True)? 發(fā)送請求實(shí)例來進(jìn)入“手動模式”。

使用 Starlette 將響應(yīng)轉(zhuǎn)發(fā)到流式處理 Web 端點(diǎn)的上下文中的示例：

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

警告
使用此“手動流式傳輸模式”時(shí)，作為開發(fā)人員，您有責(zé)任確保最終調(diào)用?Response.aclose()?。如果不這樣做，連接就會保持打開狀態(tài)，很可能導(dǎo)致資源泄漏。

流式處理請求

在發(fā)送帶有?AsyncClient?實(shí)例的流式處理請求正文時(shí)，應(yīng)使用異步字節(jié)生成器而不是字節(jié)生成器：

async def upload_bytes():
    ...  # yield byte content

await client.post(url, content=upload_bytes())

顯式transport實(shí)例

直接實(shí)例化?transport?實(shí)例時(shí)，需要使用?httpx.AsyncHTTPTransport ?。

例如：

>>> import httpx
>>> transport = httpx.AsyncHTTPTransport(retries=1)
>>> async with httpx.AsyncClient(transport=transport) as client:
>>>     ...

支持的異步環(huán)境

HTTPX 支持任一?asyncio?或?trio?作為異步環(huán)境。

它將自動檢測將這兩者中的哪一個(gè)用作套接字操作和并發(fā)基元的后端。

AsyncIO

AsyncIO 是 Python 的內(nèi)置庫，用于使用 async/await 語法編寫并發(fā)代碼。

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

trio

Trio 是一個(gè)替代異步庫，圍繞結(jié)構(gòu)化并發(fā)原則設(shè)計(jì)。

import httpx
import trio

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

trio.run(main)

重要
必須安裝?trio?軟件包才能使用 Trio 后端。

AnyIO

AnyIO 是一個(gè)異步網(wǎng)絡(luò)和并發(fā)庫，可在 ?asyncio?或 ?trio?之上工作。它與所選后端的本機(jī)庫混合（默認(rèn)為?asyncio ?）。

import httpx
import anyio

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

anyio.run(main, backend='trio')

調(diào)用 Python Web Apps

就像?httpx.Client?允許您直接調(diào)用 WSGI Web 應(yīng)用程序一樣，?httpx.AsyncClient?類允許您直接調(diào)用 ASGI Web 應(yīng)用程序。

讓我們以這個(gè)Starlette應(yīng)用程序?yàn)槔?/p>

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

我們可以直接對應(yīng)用程序發(fā)出請求，如下所示：

>>> import httpx
>>> async with httpx.AsyncClient(app=app, base_url="http://testserver") as client:
...     r = await client.get("/")
...     assert r.status_code == 200
...     assert r.text == "Hello World!"

對于一些更復(fù)雜的情況，您可能需要自定義 ASGI 傳輸。這使您可以：

• 檢查 ?500 ?錯(cuò)誤響應(yīng)，而不是通過設(shè)置 ?raise_app_exceptions=False?來引發(fā)異常。
• 通過設(shè)置?root_path ?在子路徑上掛載 ASGI 應(yīng)用程序。
• 通過設(shè)置?client ?，為請求使用給定的客戶端地址。

例如：

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

有關(guān) ?client?和 ?root_path?鍵的更多詳細(xì)信息，請參閱 ASGI 文檔。

啟動/關(guān)閉 ASGI 應(yīng)用程序

它不在 HTTPX 的范圍內(nèi)，無法觸發(fā)應(yīng)用的生存期事件。

但是，建議將 asgi-lifespan 的?LifespanManager ?與?AsyncClient ?配對使用 。