`microdot.sse` — Server-Sent Events¶

Server-Sent Events (SSE) is a simple one-way push protocol: the client opens a normal HTTP request to Content-Type: text/event-stream, and the server keeps the connection open and writes formatted events as they happen. The browser exposes the stream as a JavaScript EventSource object. Compared to WebSockets, SSE is simpler when only the server pushes; the client always requests.

The typical camera use is publishing detections / sensor readings / status updates to a phone or dashboard at the rate they happen, without the dashboard having to poll.

class SSE¶

class microdot.sse.SSE¶

The handle the route receives. Passed to handlers as the second argument when with_sse() is in use.

async send(data, event: str | None = None, event_id: str | None = None, retry: int | None = None, comment: bool = False)¶

Push a single event to the client.

data: The event payload. Strings and bytes are sent as-is. Dicts and lists are serialized to JSON. Anything else is str()-converted.
event: Optional event name – the browser’s EventSource dispatches to a JavaScript listener of the same name when this is set.
event_id: Optional event id – the browser sends it back as Last-Event-ID if the connection drops and reconnects, which lets the server resume from where it left off.
retry: Reconnection delay in seconds; the browser uses this if the connection drops.
comment: If True, the payload is sent as an SSE comment line (ignored by the browser). Useful as a keep-alive heartbeat to stop NAT timeouts from closing an otherwise-idle stream.

Module-level decorators¶

microdot.sse.with_sse(f)¶

Decorator that turns a route into an SSE endpoint. The handler receives the request and an SSE object:

from microdot import Microdot
from microdot.sse import with_sse

app = Microdot()

@app.get('/events')
@with_sse
async def events(request, sse):
    import asyncio
    while True:
        await sse.send({'temp': read_sensor()}, event='reading')
        await asyncio.sleep(1)

The handler’s lifetime equals the stream’s lifetime – it runs for as long as the client stays connected. Cancellation (the client disconnects, the server shuts down) propagates as asyncio.CancelledError, which the framework swallows.

microdot.sse.sse_response(request, event_function, *args, **kwargs)¶

Low-level entry point: returns the response tuple that with_sse() returns under the hood. Useful when an SSE endpoint needs custom headers or status codes that the decorator does not allow:

from microdot.sse import sse_response

@app.get('/events')
async def events(request):
    if not request.g.current_user:
        return 401
    async def emit(req, sse):
        await sse.send('hello')
    return sse_response(request, emit)

event_function takes (request, sse, *args, **kwargs) and uses await sse.send(...) to push events.

The SSE response sets Content-Type: text/event-stream and keeps the connection open until event_function returns or the client disconnects. Use a while True loop with an await asyncio.sleep between sends to avoid hot-looping.

microdot.sse — Server-Sent Events¶

class SSE¶

Module-level decorators¶

`microdot.sse` — Server-Sent Events¶