Faire des requêtes HTTP dans Uvicorn/FastAPI

Lors de la gestion des points de terminaison HTTP construits à l'aide de FastAPI et Uvicorn, il est courant de demander des données à des API externes. Cependant, lors du traitement de plusieurs requêtes simultanées, des erreurs telles que « Impossible de gérer le type d'événement ConnectionClosed lorsque role=SERVER et state=SEND_RESPONSE » peuvent survenir. Cela se produit car la bibliothèque client HTTP par défaut, « requests », n'est pas entièrement sécurisée pour les threads lorsqu'elle est utilisée dans un tel environnement simultané.

Pour résoudre ce problème, envisagez d'implémenter la bibliothèque client HTTP alternative appelée httpx. Il propose une API asynchrone qui permet d'éviter les problèmes de sécurité des threads dans FastAPI. L'exemple ci-dessous montre comment utiliser httpx dans FastAPI :

from fastapi import FastAPI, Request, BackgroundTask
from fastapi.responses import StreamingResponse, Response
from contextlib import asynccontextmanager
import httpx

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with httpx.AsyncClient() as client:
        yield {'client': client}

app = FastAPI(lifespan=lifespan)

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

En utilisant l'API asynchrone de httpx, vous pouvez gérer plus efficacement les requêtes HTTP dans FastAPI tout en maintenant la sécurité des threads. Vous pouvez personnaliser davantage la taille du pool de connexions à l'aide de l'argument mot-clé « limites » sur l'objet Client.