Streaming Responses

Server-sent events

import asyncio
from flasgo import EventSourceResponse, Flasgo, ServerSentEvent

app = Flasgo()


@app.get("/progress", public=True)
def progress() -> EventSourceResponse:
    async def events():
        for step in range(3):
            yield ServerSentEvent({"step": step}, event="progress", id=str(step))
            await asyncio.sleep(0.1)

    return EventSourceResponse(events())

StreamingResponse accepts an async iterable of bytes or strings. NDJSONResponse accepts async items and emits one JSON value per line. EventSourceResponse accepts ServerSentEvent objects; data is JSON encoded so embedded newlines cannot introduce another SSE field. Event names/IDs reject control characters and are length-bounded; retry values are validated integers. Optional item_model= applies a response contract to each SSE/NDJSON item’s data. Annotate the response class so OpenAPI advertises the corresponding streaming media type. Item schemas selected inside a handler are runtime configuration and are not inferred into OpenAPI.

NDJSON and byte streams

from dataclasses import dataclass

from flasgo import NDJSONResponse, StreamingResponse


@dataclass
class PublicRow:
    id: int
    name: str


@app.get("/rows", public=True)
def rows() -> NDJSONResponse:
    async def items():
        yield {"id": 1, "name": "Ada", "internal_note": "omit"}

    return NDJSONResponse(items(), item_model=PublicRow)


@app.get("/chunks", public=True)
def chunks() -> StreamingResponse:
    async def content():
        yield b"first\n"
        yield b"second\n"

    return StreamingResponse(content(), content_type="text/plain")

Use an async iterable in each case. NDJSONResponse emits application/x-ndjson, SSE emits text/event-stream, and the generic stream defaults to application/octet-stream. A return annotation of StreamingResponse documents the generic default media type; a custom content_type= chosen inside the handler is not inferred by OpenAPI.

Wrapping a streaming return type in Annotated, such as Annotated[EventSourceResponse, "progress events"], preserves its OpenAPI media type. Import Annotated from typing; its metadata does not replace item_model= for runtime item validation.

Limits and connection lifecycle

Defaults are a 64 KiB chunk/event limit, a 10-second send timeout, a 60-second wait for the next output chunk, and a one-hour maximum connection lifetime. Constructor options max_chunk_bytes, send_timeout, idle_timeout, and max_duration configure these positive finite limits. SSE emits a heartbeat every 15 seconds (heartbeat=), which counts as output for the idle timeout; the maximum lifetime still applies. EventSourceResponse raises ValueError unless heartbeat is shorter than idle_timeout, because a quiet stream would otherwise close before its first ping. Per-item JSON validation uses bounded validation budgets. Producers must also avoid allocating oversized data themselves.

Sending awaits the ASGI transport before requesting the next chunk. A disconnect, send timeout, or producer failure initiates source cleanup and closes request dependencies. If headers have already been sent, an error terminates the stream without sending a second response or appending diagnostic data. A client that disconnects mid-stream is logged as http-client-disconnected at INFO rather than as an ERROR-level response-send-failed event. Background tasks run only after successful completion. HEAD sends headers without iterating content. Response instances are single-use, have no Content-Length, and retain normal security headers. Framework session/cookie changes must occur before the handler returns; modifications made inside the producer cannot update already sent cookies. Sessions loaded before a stream opens are not continually revalidated during that connection.

Cleanup deadline

All three streaming constructors accept cleanup_timeout=10, a positive finite number of seconds separate from max_duration. It bounds waiting for producer and iterator cleanup, including cleanup after a disconnect or lifetime timeout. Keep producer finally blocks and aclose() methods cooperative with cancellation.

At the cleanup deadline, Flasgo requests cancellation and stops awaiting cancellation-resistant work. It records cleanup_timeout as the stream outcome, including after successful body delivery. This outcome does not prove resources have finished closing or that the response failed to reach ASGI. Background work can still run if sending returned normally.

The process admits at most 128 active iterator cleanups and queues at most 128 pending cleanups. Queued work is deduplicated and scheduled on its owning event loop, with its captured context, when capacity becomes available. Admission to the pending queue also records cleanup_timeout. Queue overflow raises RuntimeError and leaves cleanup retryable. These are framework bounds, not additional settings; they do not forcibly terminate application code or provide cleanup durability after a worker exits.

See the Metrics Reference for stream outcomes and byte accounting.

Browser and proxy behavior

SSE sends Cache-Control: no-store and X-Accel-Buffering: no; configure proxy buffering and connection timeouts to match. Event IDs support application-defined replay: read Last-Event-ID through Header() or Request and validate access to the requested history. Flasgo does not store events or promise automatic replay. Browser-native EventSource uses GET and does not let applications set arbitrary authorization headers; choose authentication appropriate to the client without placing credentials in URLs. CORS and CSRF continue to apply normally.

Incremental tests

Tests can read streams incrementally and disconnect deterministically:

async def test_progress() -> None:
    async with app.test_client().astream("GET", "/progress") as response:
        assert response.status_code == 200
        first_chunk = await anext(response.iter_bytes())
        assert b'event: progress\n' in first_chunk

astream is for async tests, optionally inside the client’s async lifespan context. Its one-message queue preserves backpressure. Leaving the context disconnects; a stream ending without a final body event raises an incomplete-response error. The ordinary buffered test client remains useful for finite streams.

See Testing and Dependency Lifetimes for resource cleanup and cancellation tests.