Skip to content

sse_manager

sse_manager

Server-Sent Events wire-format dataclass.

Classes

SSEEvent dataclass

SSEEvent(event, data, id=None, retry=None)

An SSE event to be sent to clients.

Functions
format
format()

Format as SSE wire format.

Source code in src/marianne/dashboard/services/sse_manager.py
def format(self) -> str:
    """Format as SSE wire format."""
    lines = []
    if self.id:
        lines.append(f"id: {self.id}")
    if self.retry:
        lines.append(f"retry: {self.retry}")
    lines.append(f"event: {self.event}")
    for line in self.data.split("\n"):
        lines.append(f"data: {line}")
    lines.append("")
    return "\n".join(lines) + "\n"