WebSockets FastAPI

WebSockets

from fastapi import WebSocket, WebSocketDisconnect
from typing import List

class ConnectionManager:
    def __init__(self):
        self.active: List[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active.append(ws)

    def disconnect(self, ws: WebSocket):
        self.active.remove(ws)

    async def broadcast(self, message: str):
        for connection in self.active:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{room}")
async def websocket_endpoint(ws: WebSocket, room: str):
    await manager.connect(ws)
    try:
        while True:
            data = await ws.receive_text()
            await manager.broadcast("[" + room + "] " + data)
    except WebSocketDisconnect:
        manager.disconnect(ws)

# Client (JavaScript)
# const ws = new WebSocket("ws://localhost:8000/ws/general");
# ws.onmessage = (e) => console.log(e.data);
# ws.send("Hello!");