當使用標準Python 請求庫在FastAPI 中發出HTTP 請求時,執行緒安全性成為並發請求中的一個問題。要有效解決此問題,請考慮使用 httpx,這是一個既提供線程安全性又提高效能的程式庫。
httpx 附帶非同步 API,讓您可以輕鬆地HTTP 請求同時高效處理多個並發任務。以下是它在FastAPI 端點中的使用範例:
from httpx import AsyncClient from fastapi import FastAPI, Request app = FastAPI() @app.on_event("startup") async def startup_event(): app.state.client = AsyncClient() @app.on_event('shutdown') async def shutdown_event(): await app.state.client.aclose() @app.get('/') async def home(request: Request): client = request.state.client req = client.build_request('GET', 'https://www.example.com') r = await client.send(req, stream=True) return StreamingResponse(r.aiter_raw(), background=BackgroundTask(r.aclose))
在此範例中:
如果不需要使用 async def 定義端點,則有必要選擇 httpx 的同步 API。這種方法維護了執行緒安全性並簡化了端點實作:
from httpx import Client from fastapi import FastAPI, Request app = FastAPI() @app.on_event("startup") def startup_event(): app.state.client = Client() @app.on_event('shutdown') async def shutdown_event(): await app.state.client.aclose() @app.get('/') def home(request: Request): client = request.state.client req = client.build_request('GET', 'https://www.example.com') try: r = client.send(req) content_type = r.headers.get('content-type') except Exception as e: content_type = 'text/plain' e = str(e) if content_type == 'application/json': return r.json() elif content_type == 'text/plain': return PlainTextResponse(content=r.text) else: return Response(content=r.content)
在此範例中,同步API 在try/ except 區塊中處理HTTP 請求,從而允許優雅地處理請求期間可能出現的任何異常。
透過利用 httpx 及其功能,您可以自信地在 FastAPI 中發出下游 HTTP 請求,無縫處理多個並發任務並確保應用程式穩定性。
以上是httpx 如何增強 FastAPI 中安全且有效率的下游 HTTP 請求?的詳細內容。更多資訊請關注PHP中文網其他相關文章!