uniqc.gateway.ws module

WebSocket event broadcaster for the gateway dashboard.

class uniqc.gateway.ws.Event(type, payload)[source]

Bases: object

One broadcast event.

payload
to_json()[source]
type
class uniqc.gateway.ws.EventBroadcaster[source]

Bases: object

ASGI-safe singleton that fans out events to all connected WebSocket clients.

Usage:

broadcaster = EventBroadcaster()
# In a route handler:
await broadcaster.broadcast(Event("task:updated", {"task_id": "abc"}))
async broadcast(event)[source]

Enqueue an event for fan-out to all connected clients.

async connect(websocket)[source]
disconnect(websocket)[source]
async emit_backend_status(backend_id, status)[source]
async emit_task_updated(task_id, status)[source]