Comment gérer efficacement les réponses en streaming dans FastAPI ?

Gestion des réponses en streaming avec FastAPI

FastAPI fournit la classe StreamingResponse pour diffuser des données aux clients pendant les appels d'API. Bien que cette fonctionnalité soit destinée au streaming de données de manière non bloquante, des problèmes peuvent survenir lors de l'utilisation d'une fonction de générateur avec des opérations de blocage ou une utilisation inappropriée.

Réponse améliorée pour le streaming de données

Pour garantir le succès streaming, considérez ce qui suit :

• Requête GET : Utilisez les requêtes GET au lieu de POST pour la récupération data.
• Identifiants sécurisés : Évitez d'envoyer des informations d'identification via la chaîne de requête ; utilisez plutôt des en-têtes ou des cookies.
• Opérations non bloquantes : Définissez les fonctions du générateur comme def (et non async def) si elles contiennent des opérations de blocage. Vous pouvez également utiliser async def avec des opérations de blocage asynchrones.
• Type de média approprié : Spécifiez un type de média différent pour la réponse (par exemple, application/json ou text/event-stream) au lieu de texte /plain, ou désactivez le reniflage MIME en définissant l'en-tête X-Content-Type-Options sur nosniff. Cela garantit que les navigateurs ne mettent pas en mémoire tampon la réponse, permettant ainsi aux données de circuler en temps réel.

Exemple de mise en œuvre

Considérez le code Python suivant :

# app.py
from fastapi import FastAPI, StreamingResponse
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def fake_data_streamer():
    for i in range(10):
        yield b'some fake data\n\n'
        await asyncio.sleep(0.5)

@app.get('/')
async def main():
    return StreamingResponse(fake_data_streamer(), media_type='text/event-stream')
    # or, use:
    '''
    headers = {'X-Content-Type-Options': 'nosniff'}
    return StreamingResponse(fake_data_streamer(), headers=headers, media_type='text/plain')
    '''

# test.py (using httpx)
import httpx

url = 'http://127.0.0.1:8000/'

with httpx.stream('GET', url) as r:
    for chunk in r.iter_raw():  # or, for line in r.iter_lines():
        print(chunk)

Ce code montre comment diffuser des données à partir d'une fonction génératrice dans une application FastAPI et les consommer à l'aide de la bibliothèque httpx.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!