`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).

microdot.sse — Server-Sent Events#

class SSE#

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

`microdot.sse` — Server-Sent Events#