`microdot.sse` --- 伺服器推送事件¶

伺服器推送事件（Server-Sent Events，SSE）是一種簡單的單向推送協定：用戶端對 Content-Type: text/event-stream 開啟一個正常的 HTTP 請求，伺服器則保持連線開啟，並在事件發生時寫出格式化的事件。瀏覽器會將該串流以 JavaScript EventSource 物件公開。相較於 WebSocket，當只有伺服器推送時 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: 選用的事件 id —— 若連線中斷並重新連線，瀏覽器會以 Last-Event-ID 將其送回，讓伺服器能從中斷處接續。
retry: 重新連線延遲（以秒計）；若連線中斷，瀏覽器會使用此值。
comment: 若為 True，酬載會作為一行 SSE 註解傳送（瀏覽器會忽略）。可作為保持連線的心跳訊號，避免 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，以避免熱迴圈空轉。

microdot.sse --- 伺服器推送事件¶

class SSE¶

模組層級裝飾器¶

`microdot.sse` --- 伺服器推送事件¶