(self, websocket: WebSocket, session_id: Optional[str] = None)
| 64 | ) |
| 65 | |
| 66 | async def connect(self, websocket: WebSocket, session_id: Optional[str] = None) -> str: |
| 67 | await websocket.accept() |
| 68 | if not session_id: |
| 69 | session_id = str(uuid.uuid4()) |
| 70 | self.active_connections[session_id] = websocket |
| 71 | self.connection_timestamps[session_id] = time.time() |
| 72 | logging.info("WebSocket connected: %s", session_id) |
| 73 | await self.send_message( |
| 74 | session_id, |
| 75 | { |
| 76 | "type": "connection", |
| 77 | "data": {"session_id": session_id, "status": "connected"}, |
| 78 | }, |
| 79 | ) |
| 80 | return session_id |
| 81 | |
| 82 | def disconnect(self, session_id: str) -> None: |
| 83 | session = self.session_store.get_session(session_id) |
no test coverage detected