fastapi判断连接是否中断

在 FastAPI 中，可以使用 Starlette 提供的 WebSocketDisconnect 异常来判断 WebSocket 连接是否中断。例如，在 WebSocket 的路由函数中，我们可以使用 try-except 语句来捕获这个异常，如下面的示例代码所示：

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections = []
    
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
    
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
    
    async def broadcast(self, data: str):
        for connection in self.active_connections:
            await connection.send_text(data)

manager = ConnectionManager()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(data)
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await websocket.close()

在上面的代码中，我们定义了一个 ConnectionManager 类来管理 WebSocket 连接，然后在 WebSocket 路由函数中使用 try-except 语句来捕获 WebSocketDisconnect 异常。如果抛出了这个异常，说明 WebSocket 连接已经中断，我们就可以调用 ConnectionManager 的 disconnect() 方法来从活动连接列表中移除这个连接，并且关闭 WebSocket 连接。