サーバー送信イベント#

Quart はストリーミングサーバー送信イベントをサポートしています。これらは WebSocket の代替として使用できます。特に、通信が一方向の場合に便利です。

サーバー送信イベントは、このヘルパークラスに示されている方法で特定の方法でエンコードする必要があります

from dataclasses import dataclass

@dataclass
class ServerSentEvent:
    data: str
    event: str | None = None
    id: int | None = None
    retry: int | None

    def encode(self) -> bytes:
        message = f"data: {self.data}"
        if self.event is not None:
            message = f"{message}\nevent: {self.event}"
        if self.id is not None:
            message = f"{message}\nid: {self.id}"
        if self.retry is not None:
            message = f"{message}\nretry: {self.retry}"
        message = f"{message}\n\n"
        return message.encode('utf-8')

ストリーミングジェネレータを返す GET ルートを使用する必要があります。以下のコード内のジェネレータ send_events は、エンコードされたサーバー送信イベントを生成する必要があります。ルート自体もクライアントが text/event-stream 応答を受け入れることを確認し、応答ヘッダを適切に設定する必要があります

from quart import abort, make_response

@app.get("/sse")
async def sse():
    if "text/event-stream" not in request.accept_mimetypes:
        abort(400)

    async def send_events():
        while True:
            data = ...  # Up to you where the events are from
            event = ServerSentEvent(data)
            yield event.encode()

    response = await make_response(
        send_events(),
        {
            'Content-Type': 'text/event-stream',
            'Cache-Control': 'no-cache',
            'Transfer-Encoding': 'chunked',
        },
    )
    response.timeout = None
    return response

デフォルトでは Quart は、サービス拒否攻撃から保護するため、長時間応答をタイムアウトします。詳細については、サービス拒否緩和を参照してください。このため、タイムアウトは無効になっています。これはグローバルに行うことができますが、他のルートを DOS に対して脆弱にする可能性があります。したがって、特定の応答でタイムアウト属性を None に設定することをお勧めします。

サーバー送信イベント#

関連項目#