Streaming

See Streaming for SSE and WebSocket examples.

format_event(envelope: Envelope) → str

Frame an envelope as one Server-Sent Event.

Parameters:

envelope – The envelope to send.

Returns:

str – "event: template\ndata: <envelope JSON>\n\n". The JSON is a single line, so one data: field is enough.

Example

>>> from anza import Envelope, format_event
>>> format_event(Envelope(slot="feed", ts=1700000000, html="<p>Hi</p>"))
'event: template\ndata: {"v":2,"slot":"feed","ts":1700000000,"html":"<p>Hi</p>"}\n\n'

format_packet(envelope: Envelope) → str

The WebSocket text message for an envelope: its compact JSON.

Parameters:

envelope – The envelope to send.

Returns:

str – Envelope.to_json().

Example

>>> from anza import Envelope, format_packet
>>> format_packet(Envelope(slot="feed", ts=1700000000, html="<p>Hi</p>"))
'{"v":2,"slot":"feed","ts":1700000000,"html":"<p>Hi</p>"}'