`microdot.sse` — Server-Sent Events¶

Server-Sent Events (SSE) är ett enkelt enkelriktat push-protokoll: klienten öppnar en vanlig HTTP-begäran mot Content-Type: text/event-stream, och servern håller anslutningen öppen och skriver formaterade händelser allteftersom de inträffar. Webbläsaren exponerar strömmen som ett JavaScript-EventSource-objekt. Jämfört med WebSockets är SSE enklare när endast servern pushar; klienten begär alltid.

Den typiska kameraanvändningen är att publicera detekteringar / sensoravläsningar / statusuppdateringar till en telefon eller instrumentpanel i den takt de inträffar, utan att instrumentpanelen behöver pollas.

class SSE¶

class microdot.sse.SSE¶

Handtaget som rutten tar emot. Skickas till hanterare som det andra argumentet när with_sse() används.

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

Pusha en enskild händelse till klienten.

data: Händelsens nyttolast. Strängar och bytes skickas som de är. Dict och list serialiseras till JSON. Allt annat konverteras med str().
event: Valfritt händelsenamn – webbläsarens EventSource skickar händelsen till en JavaScript-lyssnare med samma namn när detta är angivet.
event_id: Valfritt händelse-id – webbläsaren skickar tillbaka det som Last-Event-ID om anslutningen bryts och återansluts, vilket låter servern återuppta från där den slutade.
retry: Återanslutningsfördröjning i sekunder; webbläsaren använder detta om anslutningen bryts.
comment: Om True skickas nyttolasten som en SSE-kommentarsrad (ignoreras av webbläsaren). Användbart som hjärtslag för att hålla anslutningen vid liv och hindra NAT-timeouter från att stänga en annars inaktiv ström.

Dekoratorer på modulnivå¶

microdot.sse.with_sse(f)¶

Dekorator som gör en rutt till en SSE-slutpunkt. Hanteraren tar emot begäran och ett 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)

Hanterarens livslängd är lika med strömmens livslängd – den körs så länge klienten förblir ansluten. Avbrytande (klienten kopplar från, servern stängs av) propageras som asyncio.CancelledError, som ramverket sväljer.

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

Lågnivåingångspunkt: returnerar den svarstupel som with_sse() returnerar under huven. Användbart när en SSE-slutpunkt behöver anpassade huvuden eller statuskoder som dekoratorn inte tillåter:

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 tar (request, sse, *args, **kwargs) och använder await sse.send(...) för att pusha händelser.

SSE-svaret sätter Content-Type: text/event-stream och håller anslutningen öppen tills event_function returnerar eller klienten kopplar från. Använd en while True-loop med en await asyncio.sleep mellan sändningarna för att undvika att loopa hett.

microdot.sse — Server-Sent Events¶

class SSE¶

Dekoratorer på modulnivå¶

`microdot.sse` — Server-Sent Events¶