`microdot.sse` — Server-Sent Events¶

Server-Sent Events (SSE) ist ein einfaches Einweg-Push-Protokoll: Der Client öffnet einen normalen HTTP-Request an Content-Type: text/event-stream, und der Server hält die Verbindung offen und schreibt formatierte Ereignisse, sobald sie auftreten. Der Browser stellt den Stream als JavaScript-EventSource-Objekt bereit. Im Vergleich zu WebSockets ist SSE einfacher, wenn nur der Server pusht; der Client fordert immer an.

Die typische Verwendung bei Kameras ist die Veröffentlichung von Erkennungen / Sensormesswerten / Statusaktualisierungen an ein Telefon oder Dashboard in der Geschwindigkeit, in der sie auftreten, ohne dass das Dashboard pollen muss.

class SSE¶

class microdot.sse.SSE¶

Das Handle, das die Route erhält. Wird Handlern als zweites Argument übergeben, wenn with_sse() verwendet wird.

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

Pusht ein einzelnes Ereignis an den Client.

data: Die Ereignis-Nutzlast. Strings und Bytes werden unverändert gesendet. Dicts und Listen werden zu JSON serialisiert. Alles andere wird mit str() umgewandelt.
event: Optionaler Ereignisname – die EventSource des Browsers leitet an einen JavaScript-Listener desselben Namens weiter, wenn dies gesetzt ist.
event_id: Optionale Ereignis-ID – der Browser sendet sie als Last-Event-ID zurück, falls die Verbindung abbricht und wiederhergestellt wird, was es dem Server ermöglicht, dort fortzufahren, wo er aufgehört hat.
retry: Wiederverbindungsverzögerung in Sekunden; der Browser verwendet diese, falls die Verbindung abbricht.
comment: Wenn True, wird die Nutzlast als SSE-Kommentarzeile gesendet (vom Browser ignoriert). Nützlich als Keep-Alive-Heartbeat, um zu verhindern, dass NAT-Timeouts einen ansonsten untätigen Stream schließen.

Decorators auf Modulebene¶

microdot.sse.with_sse(f)¶

Decorator, der eine Route in einen SSE-Endpunkt verwandelt. Der Handler erhält den Request und ein SSE-Objekt:

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)

Die Lebensdauer des Handlers entspricht der Lebensdauer des Streams – er läuft so lange, wie der Client verbunden bleibt. Eine Abbruchanforderung (der Client trennt die Verbindung, der Server fährt herunter) propagiert als asyncio.CancelledError, die das Framework verschluckt.

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

Low-Level-Einstiegspunkt: gibt das Response-Tupel zurück, das with_sse() unter der Haube zurückgibt. Nützlich, wenn ein SSE-Endpunkt benutzerdefinierte Header oder Statuscodes benötigt, die der Decorator nicht zulässt:

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 nimmt (request, sse, *args, **kwargs) entgegen und verwendet await sse.send(...), um Ereignisse zu pushen.

Die SSE-Antwort setzt Content-Type: text/event-stream und hält die Verbindung offen, bis event_function zurückkehrt oder der Client die Verbindung trennt. Verwenden Sie eine while True-Schleife mit einem await asyncio.sleep zwischen den Sends, um eine Hot-Loop zu vermeiden.

microdot.sse — Server-Sent Events¶

class SSE¶

Decorators auf Modulebene¶

`microdot.sse` — Server-Sent Events¶