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