SSE (Server Sent Events)

class litestar.response.sse.ASGIStreamingSSEResponse[source]

Bases: ASGIStreamingResponse

ASGI streaming response with optional keepalive ping support for SSE.

__init__(*, ping_interval: float | None = None, **kwargs: Any) → None[source]

A low-level ASGI streaming response.

Parameters:

  • background – A background task or a list of background tasks to be executed after the response is sent.

  • content_length – The response content length.

  • cookies – The response cookies.

  • encoding – The response encoding.

  • headers – The response headers.

  • is_head_response – A boolean indicating if the response is a HEAD response.

  • iterator – An async iterator or iterable.

  • media_type – The response media type.

  • status_code – The response status code.

async send_body(send: Send, receive: Receive) → None[source]

Emit the response body, with optional keepalive pings.

class litestar.response.sse.ServerSentEvent[source]

Bases: Stream

__init__(content: str | bytes | StreamType[SSEData], *, background: BackgroundTask | BackgroundTasks | None = None, cookies: ResponseCookies | None = None, encoding: str = 'utf-8', headers: ResponseHeaders | None = None, event_type: str | None = None, event_id: int | str | None = None, retry_duration: int | None = None, comment_message: str | None = None, status_code: int | None = None, ping_interval: float | None = None) → None[source]

Initialize the response.

Parameters:

  • content – Bytes, string or a sync or async iterator or iterable.

  • background – A BackgroundTask instance or BackgroundTasks to execute after the response is finished. Defaults to None.

  • cookies – A list of Cookie instances to be set under the response Set-Cookie header.

  • encoding – The encoding to be used for the response headers.

  • headers – A string keyed dictionary of response headers. Header keys are insensitive.

  • status_code – The response status code. Defaults to 200.

  • event_type – The type of the SSE event. If given, the browser will sent the event to any ‘event-listener’ declared for it (e.g. via ‘addEventListener’ in JS).

  • event_id – The event ID. This sets the event source’s ‘last event id’.

  • retry_duration – Retry duration in milliseconds.

  • comment_message – A comment message. This value is ignored by clients and is used mostly for pinging.

  • ping_interval – Interval in seconds between keepalive pings. When set, an SSE comment (: ping) is sent at the specified interval to prevent connection timeouts from reverse proxies or clients. Defaults to None (no pings).

to_asgi_response(request: Request, *, background: BackgroundTask | BackgroundTasks | None = None, cookies: Iterable[Cookie] | None = None, headers: dict[str, str] | None = None, is_head_response: bool = False, media_type: MediaType | str | None = None, status_code: int | None = None, type_encoders: TypeEncodersMap | None = None) → ASGIResponse[source]

Create an ASGI streaming response, with optional keepalive ping support.

When ping_interval is set, returns an ASGIStreamingSSEResponse that sends periodic SSE comment pings. Otherwise delegates to the parent implementation.

Parameters:

  • request – The Request instance.

  • background – Background task(s) to be executed after the response is sent.

  • cookies – A list of cookies to be set on the response.

  • headers – Additional headers to be merged with the response headers. Response headers take precedence.

  • is_head_response – Whether the response is a HEAD response.

  • media_type – Media type for the response. If media_type is already set on the response, this is ignored.

  • status_code – Status code for the response. If status_code is already set on the response, this is ignored.

  • type_encoders – A dictionary of type encoders to use for encoding the response content.

Returns:

An ASGIStreamingResponse (or ASGIStreamingSSEResponse when ping_interval is set).

class litestar.response.sse.ServerSentEventMessage[source]

Bases: object

ServerSentEventMessage(data: ‘str | int | bytes | None’ = ‘’, event: ‘str | None’ = None, id: ‘int | str | None’ = None, retry: ‘int | None’ = None, comment: ‘str | None’ = None, sep: ‘str’ = ‘rn’)

__init__(data: str | int | bytes | None = '', event: str | None = None, id: int | str | None = None, retry: int | None = None, comment: str | None = None, sep: str = '\r\n') → None