microdot.sse — Server-Sent Events

Server-Sent Events (SSE) הוא פרוטוקול דחיפה חד-כיווני פשוט: הלקוח פותח בקשת HTTP רגילה ל-Content-Type: text/event-stream, והשרת משאיר את החיבור פתוח וכותב אירועים מעוצבים כפי שהם מתרחשים. הדפדפן חושף את הזרם כאובייקט JavaScript EventSource. בהשוואה ל-WebSockets, SSE פשוט יותר כאשר רק השרת דוחף; הלקוח תמיד מבקש.

השימוש האופייני במצלמה הוא פרסום זיהויים / קריאות חיישן / עדכוני מצב לטלפון או ללוח מחוונים בקצב שבו הם מתרחשים, מבלי שלוח המחוונים יצטרך לבצע סקירה (poll).

class SSE

class microdot.sse.SSE

הידית (handle) שהנתיב מקבל. מועברת למטפלים כארגומנט השני כאשר with_sse() בשימוש.

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

דוחף אירוע בודד אל הלקוח.

data

מטען האירוע. מחרוזות ובתים נשלחים כמות שהם. Dicts ורשימות מסודרים (serialized) ל-JSON. כל דבר אחר מומר באמצעות str().

event

שם אירוע אופציונלי – ה-EventSource של הדפדפן משגר למאזין JavaScript בעל אותו שם כאשר זה מוגדר.

event_id

מזהה אירוע אופציונלי – הדפדפן שולח אותו בחזרה כ-Last-Event-ID אם החיבור נופל ומתחבר מחדש, מה שמאפשר לשרת לחדש מהמקום שבו הפסיק.

retry

השהיית התחברות מחדש בשניות; הדפדפן משתמש בכך אם החיבור נופל.

comment

אם True, המטען נשלח כשורת הערה של SSE (שהדפדפן מתעלם ממנה). שימושי כפעימת keep-alive כדי למנוע מ-NAT timeouts לסגור זרם שאחרת היה במצב סרק.

מעטרים ברמת המודול

microdot.sse.with_sse(f)

מעטר ההופך נתיב לנקודת קצה של SSE. המטפל מקבל את הבקשה ואובייקט 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)

אורך חיי המטפל שווה לאורך חיי הזרם – הוא פועל כל עוד הלקוח נשאר מחובר. ביטול (הלקוח מתנתק, השרת נכבה) מתפשט כ-asyncio.CancelledError, שהמסגרת בולעת.

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

נקודת כניסה ברמה נמוכה: מחזירה את טאפל התגובה ש-with_sse() מחזירה מתחת למכסה המנוע. שימושי כאשר נקודת קצה של SSE זקוקה לכותרות מותאמות אישית או לקודי מצב שהמעטר אינו מאפשר:

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 מקבל (request, sse, *args, **kwargs) ומשתמש ב-await sse.send(...) כדי לדחוף אירועים.

תגובת ה-SSE קובעת Content-Type: text/event-stream ומשאירה את החיבור פתוח עד ש-event_function מחזיר או שהלקוח מתנתק. השתמש בלולאת while True עם await asyncio.sleep בין שליחות כדי להימנע מלולאה חמה (hot-looping).