Sync Client

sync

Synchronous convenience wrapper around HAClient.

The wrapper runs a dedicated event loop in a background thread and submits coroutines to it via asyncio.run_coroutine_threadsafe. This allows users in plain scripts / REPL sessions to consume the library without thinking about asyncio.

Examples:

::

from haclient import SyncHAClient

with SyncHAClient.from_url("http://localhost:8123", token=TOKEN) as ha:
    light = ha.light("kitchen")
    light.set_brightness(200)

SyncHAClient

Synchronous counterpart of HAClient.

Parameters:

Name	Type	Description	Default
`config`	`ConnectionConfig`	Resolved connection settings.	required

Source code in src/haclient/sync.py

class SyncHAClient:
    """Synchronous counterpart of `HAClient`.

    Parameters
    ----------
    config : ConnectionConfig
        Resolved connection settings.
    """

    def __init__(
        self,
        config: ConnectionConfig,
        **kwargs: Any,
    ) -> None:
        self._loop_thread = _LoopThread()

        async def _build() -> HAClient:
            return HAClient(config, **kwargs)

        self._client: HAClient = self._loop_thread.submit(_build())

    @classmethod
    def from_url(
        cls,
        base_url: str,
        *,
        token: str,
        ws_url: str | None = None,
        reconnect: bool = True,
        ping_interval: float = 30.0,
        request_timeout: float = 30.0,
        verify_ssl: bool = True,
        service_policy: ServicePolicy = "auto",
        domains: list[str] | None = None,
        load_plugins: bool = True,
    ) -> SyncHAClient:
        """Build a `SyncHAClient` from a base URL and token.

        Parameters
        ----------
        base_url : str
            Home Assistant base URL.
        token : str
            Long-lived access token.
        ws_url : str or None, optional
            Explicit WebSocket URL (derived from *base_url* when omitted).
        reconnect : bool, optional
            Whether the WebSocket should reconnect automatically.
        ping_interval : float, optional
            Seconds between WebSocket keepalive pings.
        request_timeout : float, optional
            Default timeout for individual requests.
        verify_ssl : bool, optional
            Verify TLS certificates.
        service_policy : ServicePolicy, optional
            Default service-call routing policy.
        domains : list of str or None, optional
            Restrict loaded domains. ``None`` loads all registered
            domains.
        load_plugins : bool, optional
            Discover third-party plugins via the ``haclient.domains``
            entry-point group.

        Returns
        -------
        SyncHAClient
            The configured sync client (not yet connected).
        """
        config = ConnectionConfig.from_url(
            base_url,
            token,
            ws_url=ws_url,
            reconnect=reconnect,
            ping_interval=ping_interval,
            request_timeout=request_timeout,
            verify_ssl=verify_ssl,
            service_policy=service_policy,
        )
        return cls(config, domains=domains, load_plugins=load_plugins)

    @property
    def client(self) -> HAClient:
        """Return the underlying `HAClient` instance."""
        return self._client

    def connect(self) -> None:
        """Connect the underlying client."""
        self._loop_thread.submit(self._client.connect())

    def close(self) -> None:
        """Close the underlying client and stop the background loop."""
        try:
            self._loop_thread.submit(self._client.close())
        finally:
            self._loop_thread.stop()

    def __enter__(self) -> SyncHAClient:
        """Enter the sync context manager by calling `connect`."""
        self.connect()
        return self

    def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
        """Exit the sync context manager by calling `close`."""
        self.close()

    def refresh_all(self) -> None:
        """Refresh all registered entities synchronously."""
        self._loop_thread.submit(self._client.state.refresh_all())

    def on_reconnect(
        self, handler: Callable[[], Awaitable[None] | None]
    ) -> Callable[[], Awaitable[None] | None]:
        """Register a reconnect listener.

        Parameters
        ----------
        handler : callable
            Sync or async zero-argument callable invoked after the
            WebSocket reconnects.

        Returns
        -------
        callable
            The same *handler*, returned so the method can be used as a
            decorator.
        """
        return self._client.on_reconnect(handler)

    def on_disconnect(
        self, handler: Callable[[], Awaitable[None] | None]
    ) -> Callable[[], Awaitable[None] | None]:
        """Register a disconnect listener.

        Parameters
        ----------
        handler : callable
            Sync or async zero-argument callable invoked when the
            WebSocket connection drops.

        Returns
        -------
        callable
            The same *handler*, returned so the method can be used as a
            decorator.
        """
        return self._client.on_disconnect(handler)

    # -- Domain accessors --

    def domain(self, name: str) -> _SyncDomainAccessor:
        """Return a sync accessor for *name*.

        Parameters
        ----------
        name : str
            HA domain name (e.g. ``"light"`` or a third-party domain).

        Returns
        -------
        _SyncDomainAccessor
            Blocking accessor wrapping the underlying async accessor.
        """
        return _SyncDomainAccessor(self._client.domain(name), self._loop_thread)

    def __getattr__(self, name: str) -> _SyncDomainAccessor:
        """Return a sync domain accessor for any registered domain.

        Enables ``ha.light("kitchen")``, ``ha.scene.create(...)``, etc.
        identically to `HAClient.__getattr__`, but wrapping everything
        in blocking proxies.

        Parameters
        ----------
        name : str
            The domain or accessor name.

        Returns
        -------
        _SyncDomainAccessor
            Blocking accessor wrapping the underlying async accessor.

        Raises
        ------
        AttributeError
            If *name* does not match any active domain.
        """
        # Delegate to HAClient.__getattr__ which returns a DomainAccessor
        # or raises AttributeError.
        accessor = getattr(self._client, name)
        return _SyncDomainAccessor(accessor, self._loop_thread)

client `property`

client: HAClient

Return the underlying HAClient instance.

from_url `classmethod`

from_url(base_url: str, *, token: str, ws_url: str | None = None, reconnect: bool = True, ping_interval: float = 30.0, request_timeout: float = 30.0, verify_ssl: bool = True, service_policy: ServicePolicy = 'auto', domains: list[str] | None = None, load_plugins: bool = True) -> SyncHAClient

Build a SyncHAClient from a base URL and token.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	Home Assistant base URL.	required
`token`	`str`	Long-lived access token.	required
`ws_url`	`str or None`	Explicit WebSocket URL (derived from base_url when omitted).	`None`
`reconnect`	`bool`	Whether the WebSocket should reconnect automatically.	`True`
`ping_interval`	`float`	Seconds between WebSocket keepalive pings.	`30.0`
`request_timeout`	`float`	Default timeout for individual requests.	`30.0`
`verify_ssl`	`bool`	Verify TLS certificates.	`True`
`service_policy`	`ServicePolicy`	Default service-call routing policy.	`'auto'`
`domains`	`list of str or None`	Restrict loaded domains. `None` loads all registered domains.	`None`
`load_plugins`	`bool`	Discover third-party plugins via the `haclient.domains` entry-point group.	`True`

Returns:

Type	Description
`SyncHAClient`	The configured sync client (not yet connected).

Source code in src/haclient/sync.py

@classmethod
def from_url(
    cls,
    base_url: str,
    *,
    token: str,
    ws_url: str | None = None,
    reconnect: bool = True,
    ping_interval: float = 30.0,
    request_timeout: float = 30.0,
    verify_ssl: bool = True,
    service_policy: ServicePolicy = "auto",
    domains: list[str] | None = None,
    load_plugins: bool = True,
) -> SyncHAClient:
    """Build a `SyncHAClient` from a base URL and token.

    Parameters
    ----------
    base_url : str
        Home Assistant base URL.
    token : str
        Long-lived access token.
    ws_url : str or None, optional
        Explicit WebSocket URL (derived from *base_url* when omitted).
    reconnect : bool, optional
        Whether the WebSocket should reconnect automatically.
    ping_interval : float, optional
        Seconds between WebSocket keepalive pings.
    request_timeout : float, optional
        Default timeout for individual requests.
    verify_ssl : bool, optional
        Verify TLS certificates.
    service_policy : ServicePolicy, optional
        Default service-call routing policy.
    domains : list of str or None, optional
        Restrict loaded domains. ``None`` loads all registered
        domains.
    load_plugins : bool, optional
        Discover third-party plugins via the ``haclient.domains``
        entry-point group.

    Returns
    -------
    SyncHAClient
        The configured sync client (not yet connected).
    """
    config = ConnectionConfig.from_url(
        base_url,
        token,
        ws_url=ws_url,
        reconnect=reconnect,
        ping_interval=ping_interval,
        request_timeout=request_timeout,
        verify_ssl=verify_ssl,
        service_policy=service_policy,
    )
    return cls(config, domains=domains, load_plugins=load_plugins)

connect

connect() -> None

Connect the underlying client.

Source code in src/haclient/sync.py

def connect(self) -> None:
    """Connect the underlying client."""
    self._loop_thread.submit(self._client.connect())

close

close() -> None

Close the underlying client and stop the background loop.

Source code in src/haclient/sync.py

def close(self) -> None:
    """Close the underlying client and stop the background loop."""
    try:
        self._loop_thread.submit(self._client.close())
    finally:
        self._loop_thread.stop()

enter

__enter__() -> SyncHAClient

Enter the sync context manager by calling connect.

Source code in src/haclient/sync.py

def __enter__(self) -> SyncHAClient:
    """Enter the sync context manager by calling `connect`."""
    self.connect()
    return self

exit

__exit__(exc_type: Any, exc: Any, tb: Any) -> None

Exit the sync context manager by calling close.

Source code in src/haclient/sync.py

def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
    """Exit the sync context manager by calling `close`."""
    self.close()

refresh_all

refresh_all() -> None

Refresh all registered entities synchronously.

Source code in src/haclient/sync.py

def refresh_all(self) -> None:
    """Refresh all registered entities synchronously."""
    self._loop_thread.submit(self._client.state.refresh_all())

on_reconnect

on_reconnect(handler: Callable[[], Awaitable[None] | None]) -> Callable[[], Awaitable[None] | None]

Register a reconnect listener.

Parameters:

Name	Type	Description	Default
`handler`	`callable`	Sync or async zero-argument callable invoked after the WebSocket reconnects.	required

Returns:

Type	Description
`callable`	The same handler, returned so the method can be used as a decorator.

Source code in src/haclient/sync.py

def on_reconnect(
    self, handler: Callable[[], Awaitable[None] | None]
) -> Callable[[], Awaitable[None] | None]:
    """Register a reconnect listener.

    Parameters
    ----------
    handler : callable
        Sync or async zero-argument callable invoked after the
        WebSocket reconnects.

    Returns
    -------
    callable
        The same *handler*, returned so the method can be used as a
        decorator.
    """
    return self._client.on_reconnect(handler)

on_disconnect

on_disconnect(handler: Callable[[], Awaitable[None] | None]) -> Callable[[], Awaitable[None] | None]

Register a disconnect listener.

Parameters:

Name	Type	Description	Default
`handler`	`callable`	Sync or async zero-argument callable invoked when the WebSocket connection drops.	required

Returns:

Type	Description
`callable`	The same handler, returned so the method can be used as a decorator.

Source code in src/haclient/sync.py

def on_disconnect(
    self, handler: Callable[[], Awaitable[None] | None]
) -> Callable[[], Awaitable[None] | None]:
    """Register a disconnect listener.

    Parameters
    ----------
    handler : callable
        Sync or async zero-argument callable invoked when the
        WebSocket connection drops.

    Returns
    -------
    callable
        The same *handler*, returned so the method can be used as a
        decorator.
    """
    return self._client.on_disconnect(handler)

domain

domain(name: str) -> _SyncDomainAccessor

Return a sync accessor for name.

Parameters:

Name	Type	Description	Default
`name`	`str`	HA domain name (e.g. `"light"` or a third-party domain).	required

Returns:

Type	Description
`_SyncDomainAccessor`	Blocking accessor wrapping the underlying async accessor.

Source code in src/haclient/sync.py

def domain(self, name: str) -> _SyncDomainAccessor:
    """Return a sync accessor for *name*.

    Parameters
    ----------
    name : str
        HA domain name (e.g. ``"light"`` or a third-party domain).

    Returns
    -------
    _SyncDomainAccessor
        Blocking accessor wrapping the underlying async accessor.
    """
    return _SyncDomainAccessor(self._client.domain(name), self._loop_thread)

getattr

__getattr__(name: str) -> _SyncDomainAccessor

Return a sync domain accessor for any registered domain.

Enables ha.light("kitchen"), ha.scene.create(...), etc. identically to HAClient.__getattr__, but wrapping everything in blocking proxies.

Parameters:

Name	Type	Description	Default
`name`	`str`	The domain or accessor name.	required

Returns:

Type	Description
`_SyncDomainAccessor`	Blocking accessor wrapping the underlying async accessor.

Raises:

Type	Description
`AttributeError`	If name does not match any active domain.

Source code in src/haclient/sync.py

def __getattr__(self, name: str) -> _SyncDomainAccessor:
    """Return a sync domain accessor for any registered domain.

    Enables ``ha.light("kitchen")``, ``ha.scene.create(...)``, etc.
    identically to `HAClient.__getattr__`, but wrapping everything
    in blocking proxies.

    Parameters
    ----------
    name : str
        The domain or accessor name.

    Returns
    -------
    _SyncDomainAccessor
        Blocking accessor wrapping the underlying async accessor.

    Raises
    ------
    AttributeError
        If *name* does not match any active domain.
    """
    # Delegate to HAClient.__getattr__ which returns a DomainAccessor
    # or raises AttributeError.
    accessor = getattr(self._client, name)
    return _SyncDomainAccessor(accessor, self._loop_thread)

Sync Client

sync

SyncHAClient

client property

from_url classmethod

connect

close

__enter__

__exit__

refresh_all

on_reconnect

on_disconnect

domain

__getattr__

client `property`

from_url `classmethod`

enter

exit

getattr