`microdot.websocket` --- WebSocket 支援¶

WebSocket 是建立在 HTTP 之上的雙向持久連線——在完成升級握手後，用戶端與伺服器便可在同一個 socket 上收發分幀的訊息。當相機與瀏覽器端應用程式需要全雙工通訊，而 SSE（單向推送）與一般 HTTP 輪詢無法以低成本提供時，便適合使用 WebSocket。

class WebSocket¶

class microdot.websocket.WebSocket(request)¶

路由所接收的控制代碼。當使用 with_websocket() 時，它會作為第二個引數傳遞給處理常式；請勿直接建構。

類別屬性

max_message_length: int¶: receive() 所能接受的最大酬載大小。超過此大小的訊息會引發 WebSocketError 並關閉連線。0 會停用此檢查（若您設為此值，請留意記憶體耗盡攻擊）。-1（預設值）則沿用 Request.max_body_length。

CONT: int¶: 延續幀運算碼，0x0。標記延續同一訊息中前一幀酬載的幀。Microdot 本身不會產生延續幀（每則訊息都以單一幀送出），並且在接收時會予以拒絕，但仍公開此常數，供實作自訂分幀的應用程式使用。

TEXT: int¶: 文字幀運算碼，0x1。當 data 為 str 時，send() 會自動選用此運算碼；酬載在送出時會以 UTF-8 編碼。

BINARY: int¶: 二進位幀運算碼，0x2。當 data 為 bytes / bytearray 時，send() 會自動選用此運算碼；酬載會原封不動地送出。

CLOSE: int¶: 關閉幀運算碼，0x8。由 close() 送出；接收到此幀會從 receive() 引發 WebSocketError（"Websocket connection closed"）。

PING: int¶: Ping 幀運算碼，0x9。Microdot 會在 receive() 內部自動以對應的 PONG 回應收到的 ping；應用程式通常不會察覺到它們。

PONG: int¶: Pong 幀運算碼，0xA。會自動送出以回應 PING。收到的 pong 會被靜默丟棄。

實例屬性

request: microdot.Request¶: 發起此連線的 microdot.Request——即路由處理常式所接收到的同一個物件。

closed: bool¶: 一旦 close() 執行過便為 True。可檢查此屬性以避免在清理流程中重複關閉。

方法

async receive()¶: 等待並回傳來自用戶端的下一則訊息。回傳型別與幀運算碼相符：文字幀為 str，二進位幀為 bytes。ping 幀會自動以 pong 回應；pong 幀會被靜默丟棄；close 幀則會引發 WebSocketError（"Websocket connection closed"）。

async send(data, opcode: int | None = None)¶: 將 data 傳送給用戶端。字串會以文字幀送出，位元組則以二進位幀送出；明確傳入 opcode 可覆寫此行為。

async close()¶: 送出關閉幀並將連線標記為已關閉。當 with_websocket() 的包裝器結束時會自動呼叫。

class WebSocketError¶

exception microdot.websocket.WebSocketError¶

當連線結束或協定遭到違反時，會在 WebSocket.receive() 內部引發。可在路由中用它來偵測一般的用戶端斷線:

try:
    message = await ws.receive()
except WebSocketError:
    # client closed the connection

模組層級裝飾器¶

microdot.websocket.with_websocket(f)¶

將路由轉變為 WebSocket 端點的裝飾器。處理常式會接收到請求以及一個 WebSocket 物件:

from microdot import Microdot
from microdot.websocket import with_websocket

app = Microdot()

@app.get('/echo')
@with_websocket
async def echo(request, ws):
    while True:
        msg = await ws.receive()
        await ws.send(msg)

處理常式的生命週期等同於連線的生命週期。回傳、引發例外或用戶端斷線都會乾淨地關閉 WebSocket。

async microdot.websocket.websocket_upgrade(request)¶

低階升級輔助函式。當升級應有條件地進行時——例如僅在通過授權檢查後——可在路由內部使用它:

@app.get('/private')
async def private(request):
    if not authenticate(request):
        abort(401)
    ws = await websocket_upgrade(request)
    while True:
        msg = await ws.receive()
        await ws.send(msg.upper())

回傳升級後的 WebSocket。處理常式在完成後必須回傳 microdot.Response.already_handled（或引發例外）——因為 microdot 已經接管了該 socket。

此實作支援標準的文字與二進位運算碼、ping / pong 以及乾淨的關閉。不支援分段（延續）幀——每則訊息都以單一幀送出，這對於典型的小訊息相機應用程式而言相當合適。

microdot.websocket --- WebSocket 支援¶

class WebSocket¶

class WebSocketError¶

模組層級裝飾器¶

`microdot.websocket` --- WebSocket 支援¶