服务器发送事件

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。

另请参阅

流式响应