socket_events.py
946 Bytes
"""Socket.IO event registration for the BettaFish web API."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Callable
from flask_socketio import SocketIO, emit
ProcessStatusGetter = Callable[..., dict[str, dict[str, Any]]]
@dataclass(frozen=True)
class SocketEventDependencies:
"""Runtime services needed by extracted Socket.IO event handlers."""
get_process_status: ProcessStatusGetter
def register_socketio_handlers(
socketio: SocketIO,
deps: SocketEventDependencies,
) -> None:
"""Register Socket.IO event handlers."""
@socketio.on("connect")
def handle_connect():
"""Handle a new Socket.IO client connection."""
emit("status", "Connected to Flask server")
@socketio.on("request_status")
def handle_status_request():
"""Return a fresh status snapshot to the client."""
emit("status_update", deps.get_process_status())