使用 WebSockets

要使用 WebSockets，请声明一个 WebSocket 函数而不是路由函数，如下所示：

@app.websocket('/ws')
async def ws():
    while True:
        data = await websocket.receive()
        await websocket.send(data)

websocket 是一个类似于 request 的全局变量，并共享许多相同的属性，例如 headers。

手动拒绝或接受 WebSockets

WebSocket 连接是通过接受 HTTP 升级请求创建的，但是服务器可以选择拒绝 WebSocket 请求。为此，只需从 WebSocket 函数中返回，就像您对路由函数所做的那样。

@app.websocket('/ws')
async def ws():
    if (
        websocket.authorization.username != USERNAME or
        websocket.authorization.password != PASSWORD
    ):
        return 'Invalid password', 403  # or abort(403)
    else:
        websocket.accept()  # Automatically invoked by receive or send
        ...

独立发送和接收

第一个给出的示例需要客户端发送消息才能让服务器响应。要独立发送和接收，需要独立的任务。

async def sending():
    while True:
        await websocket.send(...)

async def receiving():
    while True:
        data = await websocket.receive()
        ...

@app.websocket('/ws')
async def ws():
    producer = asyncio.create_task(sending())
    consumer = asyncio.create_task(receiving())
    await asyncio.gather(producer, consumer)

收集行至关重要，因为如果没有它，WebSocket 函数将返回并触发 Quart 发送 HTTP 响应。

检测断开连接

当客户端断开连接时，将引发 CancelledError，可以捕获该错误以处理断开连接。

@app.websocket('/ws')
async def ws():
    try:
        while True:
            data = await websocket.receive()
            await websocket.send(data)
    except asyncio.CancelledError:
        # Handle disconnection here
        raise

警告

必须重新引发 CancelledError。

关闭连接

可以通过等待 close 方法并使用适当的 WebSocket 错误代码来关闭连接。

@app.websocket('/ws')
async def ws():
    await websocket.accept()
    await websocket.close(1000)

如果 WebSocket 在被接受之前关闭，服务器将使用 403 HTTP 响应进行响应。

测试 WebSockets

要测试 WebSocket 路由，请使用 test_client，如下所示。

test_client = app.test_client()
async with test_client.websocket('/ws/') as test_websocket:
    await test_websocket.send(data)
    result = await test_websocket.receive()

如果 WebSocket 路由返回响应，test_client 将引发 WebsocketResponseError 异常，其中包含 response 属性。例如。

test_client = app.test_client()
try:
    async with test_client.websocket('/ws/') as test_websocket:
        await test_websocket.send(data)
except WebsocketResponseError as error:
    assert error.response.status_code == 401

发送和接收字节或字符串

WebSocket 协议允许使用字节或字符串进行发送，并使用帧标记来指示是哪种类型。 receive() 方法将返回 bytes 或 str，具体取决于客户端发送的内容，例如，如果客户端发送字符串，则该字符串将从方法中返回。同样，您可以发送字节或字符串。

混合 WebSocket 和 HTTP 路由

Quart 允许将路由定义为 WebSockets 和 HTTP 请求。这允许根据请求类型（WebSocket 升级或不升级）发送响应。因此，

@app.route("/ws")
async def http():
    return "A HTTP request"

@app.websocket("/ws")
async def ws():
    ...  # Use the WebSocket

如果 HTTP 定义不存在，Quart 将使用 400，错误请求，响应请求缺少的路由（而不是 404）。