`microdot.sse` — Server-Sent Events¶

Server-Sent Events (SSE) je jednoduchý jednosměrný push protokol: klient otevře běžný HTTP požadavek na Content-Type: text/event-stream a server udržuje spojení otevřené a zapisuje naformátované události tak, jak nastávají. Prohlížeč zpřístupňuje stream jako objekt JavaScriptu EventSource. Ve srovnání s WebSockety je SSE jednodušší, když pushuje pouze server; klient vždy žádá.

Typické použití na kameře je publikování detekcí / odečtů ze senzorů / aktualizací stavu na telefon nebo dashboard rychlostí, jakou nastávají, aniž by dashboard musel dotazovat.

class SSE¶

class microdot.sse.SSE¶

Handle, který routa obdrží. Předává se obslužným rutinám jako druhý argument, když se používá with_sse().

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

Odešle klientovi jedinou událost.

data: Payload události. Řetězce a bajty se odesílají tak, jak jsou. Dicty a listy se serializují do JSON. Cokoli jiného se převede pomocí str().
event: Volitelný název události – EventSource prohlížeče doručí událost JavaScriptovému posluchači stejného názvu, pokud je nastaven.
event_id: Volitelné id události – prohlížeč jej odešle zpět jako Last-Event-ID, pokud se spojení přeruší a znovu naváže, což serveru umožní pokračovat tam, kde skončil.
retry: Prodleva pro opětovné připojení v sekundách; prohlížeč ji použije, pokud se spojení přeruší.
comment: Pokud True, payload se odešle jako řádek komentáře SSE (prohlížeč jej ignoruje). Užitečné jako keep-alive heartbeat, který zabrání NAT timeoutům v uzavření jinak nečinného streamu.

Dekorátory na úrovni modulu¶

microdot.sse.with_sse(f)¶

Dekorátor, který promění routu v SSE koncový bod. Obslužná rutina obdrží požadavek a objekt SSE

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)

Životnost obslužné rutiny se rovná životnosti streamu – běží tak dlouho, dokud zůstává klient připojen. Zrušení (klient se odpojí, server se vypne) se šíří jako asyncio.CancelledError, kterou framework spolkne.

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

Nízkoúrovňový vstupní bod: vrací n-tici odpovědi, kterou with_sse() vrací pod kapotou. Užitečné, když SSE koncový bod potřebuje vlastní hlavičky nebo stavové kódy, které dekorátor neumožňuje:

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 přebírá (request, sse, *args, **kwargs) a používá await sse.send(...) k odesílání událostí.

Odpověď SSE nastaví Content-Type: text/event-stream a udržuje spojení otevřené, dokud se event_function nevrátí nebo se klient neodpojí. Použijte smyčku while True s await asyncio.sleep mezi odesíláními, abyste se vyhnuli horké smyčce.

microdot.sse — Server-Sent Events¶

class SSE¶

Dekorátory na úrovni modulu¶

`microdot.sse` — Server-Sent Events¶