microdot.sse --- أحداث مُرسَلة من الخادم (Server-Sent Events)

أحداث مُرسَلة من الخادم (SSE) هي بروتوكول دفع بسيط أحادي الاتجاه: يفتح العميل طلب HTTP عاديًا إلى Content-Type: text/event-stream، ويُبقي الخادم الاتصال مفتوحًا ويكتب أحداثًا منسقة حال وقوعها. يعرض المتصفح الدفق على هيئة كائن JavaScript EventSource. مقارنةً بـ WebSockets، يكون SSE أبسط عندما يدفع الخادم فقط؛ ويطلب العميل دائمًا.

الاستخدام النموذجي على كاميرا هو نشر عمليات الكشف / قراءات المستشعر / تحديثات الحالة إلى هاتف أو لوحة معلومات بالمعدل الذي تقع به، دون اضطرار لوحة المعلومات إلى الاستقصاء.

class SSE

class microdot.sse.SSE

المقبض الذي يتلقاه المسار. يُمرَّر إلى المعالجات كوسيط ثانٍ عند استخدام with_sse().

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

يدفع حدثًا واحدًا إلى العميل.

data

حمولة الحدث. تُرسَل السلاسل والبايتات كما هي. وتُسلسَل القواميس والقوائم بصيغة JSON. وأي شيء آخر يُحوَّل بـ str().

event

اسم حدث اختياري -- يرسل EventSource في المتصفح الحدث إلى مستمِع JavaScript بالاسم نفسه عندما يُضبَط هذا.

event_id

معرِّف حدث اختياري -- يرسله المتصفح عائدًا على هيئة Last-Event-ID إذا انقطع الاتصال وأُعيد توصيله، مما يتيح للخادم الاستئناف من حيث توقف.

retry

تأخير إعادة الاتصال بالثواني؛ يستخدمه المتصفح إذا انقطع الاتصال.

comment

إذا كانت True، فتُرسَل الحمولة على هيئة سطر تعليق SSE (يتجاهله المتصفح). مفيد كنبضة إبقاء على قيد الحياة (keep-alive) لمنع مهلات NAT من إغلاق دفق خامل لولا ذلك.

مزينات على مستوى الوحدة

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