Manages WebSocket connections for real-time communication.
| 5 | from datetime import datetime |
| 6 | |
| 7 | class ConnectionManager: |
| 8 | """Manages WebSocket connections for real-time communication.""" |
| 9 | |
| 10 | def __init__(self): |
| 11 | # Store active connections by pipeline_id |
| 12 | self.active_connections: Dict[str, List[WebSocket]] = {} |
| 13 | # Store pipeline states |
| 14 | self.pipeline_states: Dict[str, Dict] = {} |
| 15 | |
| 16 | async def connect(self, websocket: WebSocket, pipeline_id: str): |
| 17 | """Accept and store a new WebSocket connection.""" |
| 18 | await websocket.accept() |
| 19 | |
| 20 | if pipeline_id not in self.active_connections: |
| 21 | self.active_connections[pipeline_id] = [] |
| 22 | self.pipeline_states[pipeline_id] = { |
| 23 | "urls": [], |
| 24 | "schema": {}, |
| 25 | "generated_code": "", |
| 26 | "status": "connected" |
| 27 | } |
| 28 | |
| 29 | self.active_connections[pipeline_id].append(websocket) |
| 30 | |
| 31 | # Send initial state |
| 32 | await self.send_personal_message({ |
| 33 | "type": "connection", |
| 34 | "message": "Connected to pipeline", |
| 35 | "pipeline_id": pipeline_id, |
| 36 | "state": self.pipeline_states[pipeline_id] |
| 37 | }, websocket) |
| 38 | |
| 39 | def disconnect(self, websocket: WebSocket, pipeline_id: str): |
| 40 | """Remove a WebSocket connection.""" |
| 41 | if pipeline_id in self.active_connections: |
| 42 | self.active_connections[pipeline_id].remove(websocket) |
| 43 | |
| 44 | # Clean up if no more connections |
| 45 | if not self.active_connections[pipeline_id]: |
| 46 | del self.active_connections[pipeline_id] |
| 47 | |
| 48 | async def send_personal_message(self, message: Dict, websocket: WebSocket): |
| 49 | """Send a message to a specific WebSocket connection.""" |
| 50 | await websocket.send_json({ |
| 51 | **message, |
| 52 | "timestamp": datetime.utcnow().isoformat() |
| 53 | }) |
| 54 | |
| 55 | async def broadcast(self, message: Dict, pipeline_id: str): |
| 56 | """Broadcast a message to all connections for a pipeline.""" |
| 57 | if pipeline_id in self.active_connections: |
| 58 | # Create tasks for all connections |
| 59 | tasks = [] |
| 60 | for connection in self.active_connections[pipeline_id]: |
| 61 | tasks.append(connection.send_json({ |
| 62 | **message, |
| 63 | "timestamp": datetime.utcnow().isoformat() |
| 64 | })) |
no outgoing calls
no test coverage detected