FastAPI でストリーミング応答を効果的に処理するにはどうすればよいですか?

FastAPI によるストリーミング レスポンスの処理

FastAPI は、API 呼び出し中にクライアントにデータをストリーミングするための StreamingResponse クラスを提供します。この機能は非ブロッキング方式でデータをストリーミングすることを目的としていますが、ブロッキング操作や不適切な使用法でジェネレーター関数を使用すると問題が発生する可能性があります。

ストリーミング データに対する拡張応答

確実に成功するにはストリーミングの場合は、次の点を考慮してください:

• GET リクエスト: に対して POST の代わりに GET リクエストを使用します。データを取得しています。
• 安全な資格情報: クエリ文字列を介して資格情報を送信することは避けてください。代わりにヘッダーまたは Cookie を使用してください。
• 非ブロッキング操作: ジェネレーター関数にブロック操作が含まれる場合は、ジェネレーター関数を def (非同期 def ではない) として定義します。あるいは、非同期ブロッキング操作で async def を使用します。
• 適切なメディア タイプ: テキストの代わりに、応答に別のメディア タイプ (例: application/json または text/event-stream) を指定します。 /plain を使用するか、X-Content-Type-Options ヘッダーを nosniff に設定して MIME スニッフィングを無効にします。これにより、ブラウザが応答をバッファリングせず、データをリアルタイムでストリーミングできるようになります。

実装例

次の Python コードを考えてみましょう:

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

このコードは、FastAPI アプリケーションのジェネレーター関数からデータをストリーミングし、httpx ライブラリを使用してそれを使用する方法を示します。

以上がFastAPI でストリーミング応答を効果的に処理するにはどうすればよいですか?の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。