INNER CODE UNIT · Python

scan_websocket

SHAdd0WTAka/Zen-Ai-Pentest · api/main.py:790

async def scan_websocket(websocket: WebSocket, scan_id: int):
    """WebSocket for real-time scan updates"""
    await ws_manager.connect(websocket, scan_id)
    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)

            if message.get("action") == "ping":
                await websocket.send_json({"type": "pong"})

    except WebSocketDisconnect:
        ws_manager.disconnect(websocket, scan_id)


@app.websocket("/ws/notifications")
async def notifications_websocket(websocket: WebSocket):
    """WebSocket for global notifications"""

View source record →