Server Sent Events

Quart supports streaming Server Sent Events, which you may decide to use as an alternative to WebSockets - especially if the communication is one way.

Server sent events must be encoded in a specific way, as shown by this helper class:

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

To use a GET route that returns a streaming generator is required. This generator, send_events in the code below, must yield the encoded Server Sent Event. The route itself also needs to check the client will accept text/event-stream responses and set the response headers appropriately:

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 by default will timeout long responses to protect against possible denial of service attacks, see Denial Of Service mitigations. For this reason the timeout is disabled. This can be done globally, however that could make other routes DOS vulnerable, therefore the recommendation is to set the timeout attribute on the specific response to None.

See also

Streaming responses