utils.websocket

safe_send async

safe_send(ws: WebSocket, event: str, data: Any) -> bool

Safely send a message through WebSocket with error handling.

Parameters:

Name	Type	Description	Default
ws	WebSocket	The WebSocket connection	required
event	str	Event type for the message	required
data	Any	Data to send (will be converted to JSON-safe format)	required

Returns:

Type	Description
bool	True if message was sent successfully, False otherwise

Source code in src/holmes/utils/websocket.py

async def safe_send(ws: WebSocket, event: str, data: Any) -> bool:
    """
    Safely send a message through WebSocket with error handling.

    Parameters
    ----------
    ws : WebSocket
        The WebSocket connection
    event : str
        Event type for the message
    data : Any
        Data to send (will be converted to JSON-safe format)

    Returns
    -------
    bool
        True if message was sent successfully, False otherwise
    """
    if ws.client_state != WebSocketState.CONNECTED:
        logger.debug(f"Cannot send '{event}': WebSocket not connected")
        return False

    try:
        await ws.send_json({"type": event, "data": convert_for_json(data)})
        return True
    except RuntimeError as exc:
        # Connection closed during send
        logger.debug(f"Failed to send '{event}': {exc}")
        return False
    except Exception as exc:  # pragma: no cover
        logger.warning(f"Unexpected error sending '{event}': {exc}")
        return False