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())