Module

realtime.events

EventStream and SSEEvent types.

Frozen dataclasses for Server-Sent Events. The SSE handler inspects these to format the wire protocol.

2Classes

Classes

SSEEvent 6 ▼
A single Server-Sent Event.

A single Server-Sent Event.

Attributes

Name Type Description
data str —
event str | None —
id str | None —
retry int | None —

Methods

encode 0 str ▼
Serialize to SSE wire format.
def encode(self) -> str
Returns
str
Internal Methods 1 ▼
__post_init__ 0 ▼
def __post_init__(self) -> None
EventStream 4 ▼
Stream Server-Sent Events to the client. The generator yields values converted to SSE events: - `…

Stream Server-Sent Events to the client.

The generator yields values converted to SSE events:

  • str: sent as data
  • dict: JSON-serialized as data
  • Fragment: rendered via kida, sent on the htmx messagechannel unless the fragment has a target
  • SSEEvent: sent as-is

Usage::

async def stream():
    async for event in bus.subscribe():
        yield Fragment("components/item.html", item=event)
return EventStream(stream())

Attributes

Name Type Description
generator AsyncIterator[Any] —
event_type str | None —
heartbeat_interval float —

Methods

Internal Methods 1 ▼
__post_init__ 0 ▼
def __post_init__(self) -> None