Server-Sent Events

Streaming endpoints over text/event-stream. There are two entry points:

• ServerSentEventsAPIView (fastapi_views.views.sse) — a view whose abstract events action yields events; the route, framing, and OpenAPI content are generated for you.
• @sse_route (fastapi_views.views.functools) — decorate any method on a View to turn its (async) iterator of events into a streaming route.

Event models live in fastapi_views.models (BaseServerSentEvent, IdBaseServerSentEvent, AnyServerSentEvent); any object matching the fastapi_views.types.ServerSentEventType protocol (id, event, data, retry) can be yielded.

For a complete walkthrough see Server Side Events.

---

Views

ServerSentEventsAPIView

Bases: APIView, Generic[P]

API view streaming Server-Sent Events yielded by the events action.

Source code in fastapi_views/views/sse.py

class ServerSentEventsAPIView(APIView, Generic[P]):
    """API view streaming Server-Sent Events yielded by the `events` action."""

    sse_headers: ClassVar[dict[str, str]] = {
        "Cache-Control": "no-store",
        "Connection": "keep-alive",
        "X-Accel-Buffering": "no",
    }

    @classmethod
    def get_api_actions(cls, prefix: str = "") -> Generator[dict[str, Any], None, None]:
        status_code = cls.get_status_code("events", HTTP_200_OK)
        event_model = cls.get_response_schema("events") or AnyServerSentEvent
        yield cls.get_api_action(
            prefix=prefix,
            endpoint=cls.get_events_endpoint(status_code),
            methods=["GET"],
            action="events",
            status_code=status_code,
            response_model=None,
            response_class=StreamingResponse,
            responses={
                status_code: {"content": sse_openapi_content(event_model)},
            },
        )
        yield from super().get_api_actions(prefix)

    @classmethod
    def get_events_endpoint(cls, status_code: int = HTTP_200_OK) -> Endpoint:
        async def endpoint(
            self: ServerSentEventsAPIView,
            *args: P.args,
            **kwargs: P.kwargs,
        ) -> StreamingResponse:
            return StreamingResponse(
                self._serialized_events(*args, **kwargs),
                status_code=status_code,
                headers=self.sse_headers,
                media_type="text/event-stream",
            )

        cls._patch_endpoint_signature(endpoint, cls.events)
        return endpoint

    async def _serialized_events(
        self,
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> AsyncIterator[str]:
        event_schema = self.get_response_schema("events") or AnyServerSentEvent
        serializer = self.get_serializer(sse_data_annotation(event_schema))

        async for sse in self.events(*args, **kwargs):
            data = serializer.dump_json(sse.data, **self.serializer_options).decode(
                "utf-8"
            )
            yield serialize_sse(sse.id, sse.event, data, sse.retry)

    @abstractmethod
    def events(
        self, *args: P.args, **kwargs: P.kwargs
    ) -> AsyncIterator[ServerSentEventType]:
        raise NotImplementedError

---

Route decorator

fastapi_views.views.functools.sse_route(path: str = '', serializer_options: SerializerOptions | None = None, headers: dict[str, str] | None = None, **kwargs: Unpack[RouteOptions]) -> Any

Source code in fastapi_views/views/functools.py

def sse_route(
    path: str = "",
    serializer_options: SerializerOptions | None = None,
    headers: dict[str, str] | None = None,
    **kwargs: Unpack[RouteOptions],
) -> Any:
    status_code = kwargs.get("status_code", HTTP_200_OK)
    kwargs.setdefault("status_code", HTTP_200_OK)
    kwargs.setdefault("methods", ["GET"])
    response_model = kwargs.pop("response_model", None) or AnyServerSentEvent
    data_serializer: TypeAdapter[Any] = TypeAdapter(sse_data_annotation(response_model))
    kwargs.update(
        {
            "response_model": None,
            "response_class": StreamingResponse,
            "responses": {
                status_code: {"content": sse_openapi_content(response_model)},
            },
        },
    )
    if headers is None:
        headers = {
            "Cache-Control": "no-store",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        }

    def wrapper(
        func: Callable[
            Concatenate[V, _P],
            AsyncIterator[ServerSentEventType],
        ]
        | Callable[Concatenate[V, _P], Iterator[ServerSentEventType]],
    ) -> Callable[Concatenate[V, _P], Awaitable[StreamingResponse]]:

        @functools.wraps(func)
        async def wrapped(
            self: V,
            *args: _P.args,
            **kwargs: _P.kwargs,
        ) -> StreamingResponse:
            async_iterator = _wrapped_events(
                func(self, *args, **kwargs),
                data_serializer,
                **(serializer_options or {}),
            )
            return StreamingResponse(
                async_iterator,
                status_code=status_code,
                media_type="text/event-stream",
                headers=headers,
            )

        return route(path, **kwargs)(wrapped)

    return wrapper

---

Streaming event models

Discriminated lifecycle events for long-running streamed operations, in fastapi_views.models.streaming.

Universal responses schemas for streaming results, for example with SSE, loosely inspired by OpenAI responses API

TimestampData

Bases: BaseSchema

Payload base carrying the UTC timestamp of the event.

Source code in fastapi_views/models/streaming.py

class TimestampData(BaseSchema):
    """Payload base carrying the UTC timestamp of the event."""

    timestamp: int = Field(default_factory=timestamp, description="UTC timestamp")

StartedData

Bases: TimestampData

Payload of :class:ResponseStarted.

Source code in fastapi_views/models/streaming.py

class StartedData(TimestampData):
    """Payload of :class:`ResponseStarted`."""

    type: Literal["response.started"] = "response.started"

ResponseStarted

Bases: _BaseEvent

Event emitted when the response stream starts.

Source code in fastapi_views/models/streaming.py

class ResponseStarted(_BaseEvent):
    """Event emitted when the response stream starts."""

    event: Literal["response.started"] = "response.started"
    data: StartedData

    @classmethod
    def new(cls) -> Self:
        return cls(data=StartedData())

ErrorData

Bases: BaseSchema

Payload of :class:ResponseError.

Source code in fastapi_views/models/streaming.py

class ErrorData(BaseSchema):
    """Payload of :class:`ResponseError`."""

    type: Literal["response.error"] = "response.error"
    error: str

ResponseError

Bases: _BaseEvent

Event emitted when the stream fails with an error.

Source code in fastapi_views/models/streaming.py

class ResponseError(_BaseEvent):
    """Event emitted when the stream fails with an error."""

    event: Literal["response.error"] = "response.error"
    data: ErrorData

    @classmethod
    def new(cls, error: str) -> Self:
        return cls(data=ErrorData(error=error))

ResultData

Bases: BaseSchema, Generic[T]

Payload of :class:ResponseResult: a batch of result items.

Source code in fastapi_views/models/streaming.py

class ResultData(BaseSchema, Generic[T]):
    """Payload of :class:`ResponseResult`: a batch of result items."""

    type: Literal["response.result"] = "response.result"
    items: list[T] = Field(description="List of results")
    index: int | None = Field(None, description="Optional result index (page number)")
    total_results: int | None = Field(
        None, description="Optional total number of results to expect"
    )

ResponseResult

Bases: _BaseEvent, Generic[T]

Event carrying a batch of result items.

Source code in fastapi_views/models/streaming.py

class ResponseResult(_BaseEvent, Generic[T]):
    """Event carrying a batch of result items."""

    event: Literal["response.result"] = "response.result"
    data: ResultData[T]

    @classmethod
    def new(
        cls,
        items: list[T],
        *,
        index: int | None = None,
        total_results: int | None = None,
    ) -> Self:
        return cls.model_validate(
            {
                "data": {
                    "items": items,
                    "index": index,
                    "total_results": total_results,
                },
            },
        )

FinishedData

Bases: TimestampData

Payload of :class:ResponseFinished.

Source code in fastapi_views/models/streaming.py

class FinishedData(TimestampData):
    """Payload of :class:`ResponseFinished`."""

    type: Literal["response.finished"] = "response.finished"
    duration_s: NonNegativeInt | None = Field(
        None, description="Optional duration in seconds"
    )

ResponseFinished

Bases: _BaseEvent

Event emitted when the stream completes successfully.

Source code in fastapi_views/models/streaming.py

class ResponseFinished(_BaseEvent):
    """Event emitted when the stream completes successfully."""

    event: Literal["response.finished"] = "response.finished"
    data: FinishedData

    @classmethod
    def new(cls, duration_s: int | None = None) -> Self:
        return cls(data=FinishedData(duration_s=duration_s))

CancelledData

Bases: TimestampData

Payload of :class:ResponseCancelled.

Source code in fastapi_views/models/streaming.py

class CancelledData(TimestampData):
    """Payload of :class:`ResponseCancelled`."""

    type: Literal["response.cancelled"] = "response.cancelled"

ResponseCancelled

Bases: _BaseEvent

Event emitted when the stream is cancelled.

Source code in fastapi_views/models/streaming.py

class ResponseCancelled(_BaseEvent):
    """Event emitted when the stream is cancelled."""

    event: Literal["response.cancelled"] = "response.cancelled"
    data: CancelledData

    @classmethod
    def new(cls) -> Self:
        return cls(data=CancelledData())

timestamp() -> int

Current UTC time as a unix timestamp in whole seconds.

Source code in fastapi_views/models/streaming.py

def timestamp() -> int:
    """Current UTC time as a unix timestamp in whole seconds."""
    return int(datetime.now(tz=timezone.utc).timestamp())