`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#