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

WebSocket 是建立在 HTTP 之上的双向持久连接——在完成升级握手之后，客户端和服务器可以在同一个套接字上收发分帧的消息。当摄像头与浏览器端的应用需要全双工通信，而 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 已经接管了该套接字。

该实现支持标准的文本和二进制操作码、ping / pong 以及干净的关闭。不支持分片（续帧）帧——每条消息都作为单一帧发送，这对于典型的小消息摄像头应用而言是合适的。

microdot.websocket --- WebSocket 支持¶

class WebSocket¶

class WebSocketError¶

模块级装饰器¶

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