ストリーミングレスポンス#
Quartは、一度にまとめて受信するのではなく、クライアントにストリーミングすることを目的としたレスポンスをサポートしています。リクエストデータのストリーミングに関心がある場合は、リクエストボディの消費を参照するか、双方向ストリーミングの場合はWebSocketsの使用を参照してください。
レスポンスをストリーミングするには、ビュー関数はバイトをyieldする非同期ジェネレータを返す必要があります。このジェネレータは、通常のステータスコードとヘッダーと共に返すことができます。たとえば、毎秒時間をストリーミングするには、
@app.route('/')
async def stream_time():
async def async_generator():
time = datetime.isoformat()
yield time.encode()
return async_generator(), 200, {'X-Something': 'value'}
コンテキストを使用#
ストリーミング中にrequestコンテキストを使用する場合は、quart.helpers.stream_with_context()デコレータを使用する必要があります。
@app.route('/')
async def stream_time():
@stream_with_context
async def async_generator():
time = datetime.isoformat()
yield time.encode()
return async_generator(), 200, {'X-Something': 'value'}
タイムアウト#
Quartはデフォルトで、サービス拒否攻撃の可能性に対する保護として、長いレスポンスのタイムアウトを設定します。サービス拒否攻撃の軽減策を参照してください。これは、無期限のストリームなど、ストリーミングレスポンスには望ましくない場合があります。タイムアウトはグローバルに無効にできますが、これにより他のルートがDoS攻撃に対して脆弱になる可能性があるため、特定のレスポンスのタイムアウト属性をNoneに設定することをお勧めします。
from quart import make_response
@app.route('/sse')
async def stream_time():
...
response = await make_response(async_generator())
response.timeout = None # No timeout for this route
return response
テスト#
テストクライアントget()および関連メソッドは、ストリーミングされたレスポンス全体を収集します。ルートが実際にレスポンスをストリーミングするかどうかをテストする場合、またはクライアントが切断するまでストリーミングするルートをテストする場合は、request()メソッドを使用する必要があります。
async def test_stream() -> None:
test_client = app.test_client()
async with test_client.request(..) as connection:
data = await connection.receive()
assert data ...
assert connection.status_code == 200
...
await connection.disconnect() # For infinite streams