Skip to content

Service Caller

Routes raw HA service invocations between the WebSocket and REST transports according to a ServicePolicy ("ws", "rest", or "auto"). Most users should call services through the domain entity methods (e.g. ha.light("kitchen").set_brightness(...)) and treat ServiceCaller.call as an escape hatch — see Service routing (advanced).

services

ServiceCaller — explicit WS / REST routing for service invocations.

The previous implementation chose between WebSocket and REST inside a private helper on HAClient. That coupled transport policy to the client class and made the choice invisible to callers. ServiceCaller makes the policy first-class:

  • The default policy is fixed at construction time.
  • Each call may override the policy via the prefer parameter.
Policies
  • "ws" — always use the WebSocket. Fails with ConnectionClosedError if the WS is not connected.
  • "rest" — always use REST.
  • "auto" — prefer WebSocket when connected, otherwise fall back to REST.

ServiceCaller

Route call_service invocations between WebSocket and REST.

Parameters:

Name Type Description Default
rest RestPort

Adapter implementing the REST transport.

 required
ws WebSocketPort

Adapter implementing the WebSocket transport.

 required
default_policy ServicePolicy

Default routing policy used when prefer is omitted.

 'auto'
Source code in src/haclient/core/services.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
class ServiceCaller:
    """Route ``call_service`` invocations between WebSocket and REST.

    Parameters
    ----------
    rest : RestPort
        Adapter implementing the REST transport.
    ws : WebSocketPort
        Adapter implementing the WebSocket transport.
    default_policy : ServicePolicy
        Default routing policy used when ``prefer`` is omitted.
    """

    def __init__(
        self,
        rest: RestPort,
        ws: WebSocketPort,
        *,
        default_policy: ServicePolicy = "auto",
    ) -> None:
        self._rest = rest
        self._ws = ws
        self._default_policy = default_policy

    @property
    def default_policy(self) -> ServicePolicy:
        """Return the default routing policy."""
        return self._default_policy

    @property
    def ws(self) -> WebSocketPort:
        """Return the bound `WebSocketPort`.

        Exposed for advanced domain code that needs to send custom
        WebSocket commands (e.g. ``timer/create``) which are neither
        services nor event subscriptions.
        """
        return self._ws

    @property
    def rest(self) -> RestPort:
        """Return the bound `RestPort`."""
        return self._rest

    async def call(
        self,
        domain: str,
        service: str,
        data: dict[str, Any] | None = None,
        *,
        prefer: ServicePolicy | None = None,
    ) -> Any:
        """Invoke a Home Assistant service.

        Parameters
        ----------
        domain : str
            The service domain.
        service : str
            The service name.
        data : dict or None, optional
            Service data payload.
        prefer : ServicePolicy or None, optional
            Per-call policy override. ``None`` uses the default policy.

        Returns
        -------
        Any
            The result returned by Home Assistant.

        Raises
        ------
        ConnectionClosedError
            If ``prefer="ws"`` is requested but the WebSocket is not connected.
        """
        policy: ServicePolicy = prefer if prefer is not None else self._default_policy
        if policy == "rest":
            return await self._rest.call_service(domain, service, data)
        if policy == "ws":
            if not self._ws.connected:
                raise ConnectionClosedError(
                    "Service call requested via WebSocket but WS is not connected"
                )
            return await self._call_ws(domain, service, data)
        # auto
        if self._ws.connected:
            return await self._call_ws(domain, service, data)
        return await self._rest.call_service(domain, service, data)

    async def _call_ws(
        self,
        domain: str,
        service: str,
        data: dict[str, Any] | None,
    ) -> Any:
        """Send the WebSocket ``call_service`` payload."""
        payload: dict[str, Any] = {
            "type": "call_service",
            "domain": domain,
            "service": service,
        }
        if data:
            payload["service_data"] = data
        return await self._ws.send_command(payload)

default_policy property 

default_policy: ServicePolicy

Return the default routing policy.

ws property 

ws: WebSocketPort

Return the bound WebSocketPort.

Exposed for advanced domain code that needs to send custom WebSocket commands (e.g. timer/create) which are neither services nor event subscriptions.

rest property 

rest: RestPort

Return the bound RestPort.

call async 

call(domain: str, service: str, data: dict[str, Any] | None = None, *, prefer: ServicePolicy | None = None) -> Any

Invoke a Home Assistant service.

Parameters:

Name Type Description Default
domain str

The service domain.

 required
service str

The service name.

 required
data dict or None

Service data payload.

 None
prefer ServicePolicy or None

Per-call policy override. None uses the default policy.

 None

Returns:

Type Description
Any

The result returned by Home Assistant.

Raises:

Type Description
ConnectionClosedError

If prefer="ws" is requested but the WebSocket is not connected.
Source code in src/haclient/core/services.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def call(
    self,
    domain: str,
    service: str,
    data: dict[str, Any] | None = None,
    *,
    prefer: ServicePolicy | None = None,
) -> Any:
    """Invoke a Home Assistant service.

    Parameters
    ----------
    domain : str
        The service domain.
    service : str
        The service name.
    data : dict or None, optional
        Service data payload.
    prefer : ServicePolicy or None, optional
        Per-call policy override. ``None`` uses the default policy.

    Returns
    -------
    Any
        The result returned by Home Assistant.

    Raises
    ------
    ConnectionClosedError
        If ``prefer="ws"`` is requested but the WebSocket is not connected.
    """
    policy: ServicePolicy = prefer if prefer is not None else self._default_policy
    if policy == "rest":
        return await self._rest.call_service(domain, service, data)
    if policy == "ws":
        if not self._ws.connected:
            raise ConnectionClosedError(
                "Service call requested via WebSocket but WS is not connected"
            )
        return await self._call_ws(domain, service, data)
    # auto
    if self._ws.connected:
        return await self._call_ws(domain, service, data)
    return await self._rest.call_service(domain, service, data)