WebSocket status stream
(websocket: WebSocket)
| 136 | |
| 137 | @router.websocket("/ws/status") |
| 138 | async def websocket_status(websocket: WebSocket): |
| 139 | """WebSocket status stream""" |
| 140 | await websocket.accept() |
| 141 | |
| 142 | try: |
| 143 | while True: |
| 144 | # Send status every second |
| 145 | status = crawler_manager.get_status() |
| 146 | await websocket.send_json(status) |
| 147 | await asyncio.sleep(1) |
| 148 | except WebSocketDisconnect: |
| 149 | pass |
| 150 | except Exception: |
| 151 | pass |
nothing calls this directly
no test coverage detected