| from fastapi import WebSocket, WebSocketDisconnect |
| from typing import Dict, Set |
| import json |
| import asyncio |
|
|
| class CollaborationManager: |
| def __init__(self): |
| self.active_connections: Dict[str, Set[WebSocket]] = {} |
| |
| async def connect(self, session_id: str, websocket: WebSocket): |
| await websocket.accept() |
| if session_id not in self.active_connections: |
| self.active_connections[session_id] = set() |
| self.active_connections[session_id].add(websocket) |
| |
| async def disconnect(self, session_id: str, websocket: WebSocket): |
| self.active_connections[session_id].remove(websocket) |
| |
| async def broadcast(self, session_id: str, message: dict): |
| if session_id in self.active_connections: |
| for connection in self.active_connections[session_id]: |
| await connection.send_json(message) |
|
|
| collaboration_manager = CollaborationManager() |
|
|
