`microdot.sse` — Server-Sent Events#

Server-Sent Events (SSE) is een eenvoudig eenrichtings-pushprotocol: de client opent een normaal HTTP-request naar Content-Type: text/event-stream, en de server houdt de verbinding open en schrijft geformatteerde events zodra ze plaatsvinden. De browser stelt de stream beschikbaar als een JavaScript EventSource-object. Vergeleken met WebSockets is SSE eenvoudiger wanneer alleen de server pusht; de client doet altijd het verzoek.

Het typische cameragebruik is het publiceren van detecties / sensoraflezingen / statusupdates naar een telefoon of dashboard met de snelheid waarmee ze plaatsvinden, zonder dat het dashboard hoeft te pollen.

class SSE#

class microdot.sse.SSE#

De handle die de route ontvangt. Wordt als tweede argument aan handlers doorgegeven wanneer with_sse() in gebruik is.

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

Push een enkel event naar de client.

data: De event-payload. Strings en bytes worden as-is verzonden. Dicts en lists worden geserialiseerd naar JSON. Al het andere wordt geconverteerd met str().
event: Optionele eventnaam – de EventSource van de browser dispatcht naar een JavaScript-listener met dezelfde naam wanneer dit is ingesteld.
event_id: Optionele event-id – de browser stuurt deze terug als Last-Event-ID als de verbinding wegvalt en opnieuw verbinding maakt, waardoor de server kan hervatten waar hij was gebleven.
retry: Vertraging voor opnieuw verbinden in seconden; de browser gebruikt dit als de verbinding wegvalt.
comment: Indien True wordt de payload verzonden als een SSE-commentaarregel (genegeerd door de browser). Handig als keep-alive-hartslag om te voorkomen dat NAT-timeouts een verder inactieve stream sluiten.

Decorators op moduleniveau#

microdot.sse.with_sse(f)#

Decorator die een route omzet in een SSE-endpoint. De handler ontvangt het request en een 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)

De levensduur van de handler is gelijk aan de levensduur van de stream – hij draait zolang de client verbonden blijft. Annulering (de client verbreekt de verbinding, de server sluit af) propageert als asyncio.CancelledError, die het framework opvangt.

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

Laag-niveau-ingang: geeft de respons-tuple terug die with_sse() onder de motorkap teruggeeft. Handig wanneer een SSE-endpoint aangepaste headers of statuscodes nodig heeft die de decorator niet toestaat:

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 neemt (request, sse, *args, **kwargs) en gebruikt await sse.send(...) om events te pushen.

De SSE-respons stelt Content-Type: text/event-stream in en houdt de verbinding open totdat event_function terugkeert of de client de verbinding verbreekt. Gebruik een while True-lus met een await asyncio.sleep tussen verzendingen om hot-looping te voorkomen.

microdot.sse — Server-Sent Events#

class SSE#

Decorators op moduleniveau#

`microdot.sse` — Server-Sent Events#