Index

API functionality for interacting with Home Assistant.

This module provides clean access to the API classes for making HTTP requests, managing WebSocket connections, and handling entity states.

Api

Bases: Resource

API service for interacting with Home Assistant.

This service provides methods to interact with the Home Assistant API, including making REST requests, managing WebSocket connections, and handling entity states.

Source code in src/hassette/api/api.py

class Api(Resource):
    """API service for interacting with Home Assistant.

    This service provides methods to interact with the Home Assistant API, including making REST requests,
    managing WebSocket connections, and handling entity states.
    """

    sync: ApiSyncFacade
    """Synchronous facade for the API service."""

    helpers: HelperClient
    """Client for CRUD operations on Home Assistant helper entities."""

    _api_service: "ApiResource"
    """Internal API service instance."""

    def __init__(self, hassette: "Hassette", *, parent: Resource | None = None) -> None:
        super().__init__(hassette, parent=parent)
        self._api_service = self.hassette.api_service
        # helpers must be constructed before sync — ApiSyncFacade.__init__ reads
        # self._api.helpers to wire its own nested HelperClientSyncFacade.
        self.helpers = self.add_child(HelperClient, api=self)
        self.sync = self.add_child(ApiSyncFacade, api=self)

    async def on_initialize(self) -> None:
        mark_ready(self, reason="API initialized")

    @property
    def config_log_level(self) -> LOG_LEVEL_TYPE:
        """Return the log level from the config for this resource."""
        return self.hassette.config.logging.api

    async def ws_send_and_wait(self, **data: Any) -> Any:
        """Send a WebSocket message and wait for a response."""
        return await self._api_service.ws_conn.send_and_wait(**data)

    async def ws_send_json(self, **data: Any) -> None:
        """Send a WebSocket message without waiting for a response."""
        await self._api_service.ws_conn.send_json(**data)

    async def rest_request(
        self,
        method: str,
        url: str,
        params: dict[str, Any] | None = None,
        data: dict[str, Any] | None = None,
        suppress_error_message: bool = False,
        **kwargs: Any,
    ) -> aiohttp.ClientResponse:
        """Make a REST request to the Home Assistant API.

        Args:
            method: The HTTP method to use (e.g., "GET", "POST").
            url: The URL endpoint for the request.
            params: Query parameters for the request.
            data: JSON payload for the request.
            suppress_error_message: Whether to suppress error messages.

        Returns:
            The response from the API.
        """
        return await self._api_service.rest_request(
            method, url, params=params, data=data, suppress_error_message=suppress_error_message, **kwargs
        )

    async def get_rest_request(
        self, url: str, params: dict[str, Any] | None = None, **kwargs: Any
    ) -> aiohttp.ClientResponse:
        """Make a GET request to the Home Assistant API.

        Args:
            url: The URL endpoint for the request.
            params: Query parameters for the request.
            kwargs: Additional keyword arguments to pass to the request.

        Returns:
            The response from the API.
        """
        return await self.rest_request("GET", url, params=params, **kwargs)

    async def post_rest_request(
        self, url: str, data: dict[str, Any] | None = None, **kwargs: Any
    ) -> aiohttp.ClientResponse:
        """Make a POST request to the Home Assistant API.

        Args:
            url: The URL endpoint for the request.
            data: JSON payload for the request.
            kwargs: Additional keyword arguments to pass to the request.

        Returns:
            The response from the API.
        """
        return await self.rest_request("POST", url, data=data, **kwargs)

    async def delete_rest_request(self, url: str, **kwargs: Any) -> aiohttp.ClientResponse:
        """Make a DELETE request to the Home Assistant API.

        Args:
            url: The URL endpoint for the request.
            kwargs: Additional keyword arguments to pass to the request.

        Returns:
            The response from the API.
        """
        return await self.rest_request("DELETE", url, **kwargs)

    async def get_states_raw(self) -> list["HassStateDict"]:
        """Get all entities in Home Assistant as raw dictionaries.

        Returns:
            A list of states as dictionaries.
        """
        val = await self.ws_send_and_wait(type="get_states")
        return _expect_list(val, "get_states")

    async def get_states(self) -> list["BaseState"]:
        """Get all entities in Home Assistant, converted to their appropriate state types.

        If a state fails to convert, it is skipped with an error logged. If there is no registered
        state class for a domain, the generic BaseState is used.

        Returns:
            A list of states, converted to their appropriate state types.
        """
        val = await self.get_states_raw()

        self.logger.debug("Converting states to specific state types")
        converted: list[BaseState] = []

        for raw_state in val:
            # the conversion method will handle logging any conversion errors
            with suppress(UnableToConvertStateError):
                state = self.hassette.state_registry.try_convert_state(raw_state)
                converted.append(state)

        return converted

    async def get_config(self) -> dict[str, Any]:
        """Get the Home Assistant configuration.

        Returns:
            The configuration data.
        """
        val = await self.ws_send_and_wait(type="get_config")
        return _expect_dict(val, "get_config")

    async def get_services(self) -> dict[str, Any]:
        """Get the available services in Home Assistant.

        Returns:
            The services data.
        """
        val = await self.ws_send_and_wait(type="get_services")
        return _expect_dict(val, "get_services")

    async def get_panels(self) -> dict[str, Any]:
        """Get the available panels in Home Assistant.

        Returns:
            The panels data.
        """
        val = await self.ws_send_and_wait(type="get_panels")
        return _expect_dict(val, "get_panels")

    def fire_event(
        self,
        event_type: str,
        event_data: dict[str, Any] | None = None,
    ) -> "Coroutine[Any, Any, dict[str, Any]]":
        """Fire a custom event in Home Assistant.

        Must be awaited — a forgotten ``await`` is reported per ``forgotten_await_behavior`` (default: warn).

        Args:
            event_type: The type of the event to fire (e.g., "custom_event").
            event_data: Additional data to include with the event.

        Returns:
            The response from Home Assistant.
        """
        # Cheap path: no DB-record telemetry on api fire-and-forget methods
        # (unlike bus/scheduler listeners) — only warning attribution needs the location here.
        source_location = capture_source_location()
        # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
        return guard_await(
            self._fire_event(event_type, event_data),
            owner=self.parent,
            source_location=source_location,
            method_name="fire_event",
        )

    async def _fire_event(
        self,
        event_type: str,
        event_data: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Async body for fire_event."""
        event_data = event_data or {}

        data = {"type": "fire_event", "event_type": event_type, "event_data": event_data}
        if not event_data:
            data.pop("event_data")

        return await self.ws_send_and_wait(**data)

    # Overload order is load-bearing — Pyright matches top-to-bottom, first hit wins.
    # The None-returning (return_response not True) overload MUST come first so a call
    # that omits return_response resolves to it. The True overload needs a default
    # (target already has one, and a non-default arg can't follow a default arg), but
    # that default never makes it the chosen overload for an omitted return_response,
    # because the None overload above matches first. Reordering or dropping the default
    # silently flips the inferred return type — covered by tests/pyright_probes.
    @overload
    def call_service(
        self,
        domain: str,
        service: str,
        target: dict[str, str] | dict[str, list[str]] | None = None,
        return_response: typing.Literal[False] | None = None,
        **data: Any,
    ) -> "Coroutine[Any, Any, None]": ...

    @overload
    def call_service(
        self,
        domain: str,
        service: str,
        target: dict[str, str] | dict[str, list[str]] | None = None,
        return_response: Literal[True] = True,
        **data: Any,
    ) -> "Coroutine[Any, Any, ServiceResponse]": ...

    def call_service(
        self,
        domain: str,
        service: str,
        target: dict[str, str] | dict[str, list[str]] | None = None,
        return_response: bool | None = False,
        **data: Any,
    ) -> "Coroutine[Any, Any, ServiceResponse | None]":
        """Call a Home Assistant service.

        Must be awaited — a forgotten ``await`` is reported per ``forgotten_await_behavior`` (default: warn).

        Args:
            domain: The domain of the service (e.g., "light").
            service: The name of the service to call (e.g., "turn_on").
            target: Target entity IDs or areas.
            return_response: Whether to return the response from Home Assistant. Defaults to False.
            **data: Additional data to send with the service call.

        Returns:
            ServiceResponse | None: The response from Home Assistant if return_response is True. Otherwise None.
        """
        # Cheap path — see fire_event (same rationale for all api methods)
        source_location = capture_source_location()
        # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
        return guard_await(
            self._call_service(domain, service, target, return_response, **data),
            owner=self.parent,
            source_location=source_location,
            method_name="call_service",
        )

    async def _call_service(
        self,
        domain: str,
        service: str,
        target: dict[str, str] | dict[str, list[str]] | None = None,
        return_response: bool | None = False,
        **data: Any,
    ) -> ServiceResponse | None:
        """Async body for call_service."""
        payload = {
            "type": "call_service",
            "domain": domain,
            "service": service,
            "target": target,
            "return_response": return_response,
        }

        payload = {k: v for k, v in payload.items() if v is not None}
        data = {k: v for k, v in data.items() if v is not None}

        if data:
            self.logger.debug("Adding extra data to service call: %s", data)
            payload["service_data"] = data

        if return_response:
            resp = await self.ws_send_and_wait(**payload)
            return ServiceResponse(**resp)

        await self.ws_send_json(**payload)
        return None

    def turn_on(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> "Coroutine[Any, Any, None]":
        """Turn on a specific entity in Home Assistant.

        Must be awaited — a forgotten ``await`` is reported per ``forgotten_await_behavior`` (default: warn).

        Args:
            entity_id: The ID of the entity to turn on (e.g., "light.office").
            domain: The domain to use for the service call. Defaults to the entity's domain,
                derived from ``entity_id`` (e.g., ``"light.office"`` derives ``"light"``). Pass
                explicitly to override — for example, ``domain="homeassistant"`` to use the
                generic (deprecated) service.
            **data: Additional service data forwarded to the service call.

        """
        entity_id = str(entity_id)
        if domain is None:
            domain = entity_id.split(".", 1)[0]
        return self.call_service(domain=domain, service="turn_on", target={"entity_id": entity_id}, **data)

    def turn_off(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> "Coroutine[Any, Any, None]":
        """Turn off a specific entity in Home Assistant.

        Must be awaited — a forgotten ``await`` is reported per ``forgotten_await_behavior`` (default: warn).

        Args:
            entity_id: The ID of the entity to turn off (e.g., "light.office").
            domain: The domain to use for the service call. Defaults to the entity's domain,
                derived from ``entity_id`` (e.g., ``"light.office"`` derives ``"light"``). Pass
                explicitly to override — for example, ``domain="homeassistant"`` to use the
                generic (deprecated) service.
            **data: Additional service data forwarded to the service call.

        """
        entity_id = str(entity_id)
        if domain is None:
            domain = entity_id.split(".", 1)[0]
        return self.call_service(domain=domain, service="turn_off", target={"entity_id": entity_id}, **data)

    def toggle(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> "Coroutine[Any, Any, None]":
        """Toggle a specific entity in Home Assistant.

        Must be awaited — a forgotten ``await`` is reported per ``forgotten_await_behavior`` (default: warn).

        Args:
            entity_id: The ID of the entity to toggle (e.g., "light.office").
            domain: The domain to use for the service call. Defaults to the entity's domain,
                derived from ``entity_id`` (e.g., ``"light.office"`` derives ``"light"``). Pass
                explicitly to override — for example, ``domain="homeassistant"`` to use the
                generic (deprecated) service.
            **data: Additional service data forwarded to the service call.

        """
        entity_id = str(entity_id)
        if domain is None:
            domain = entity_id.split(".", 1)[0]
        return self.call_service(domain=domain, service="toggle", target={"entity_id": entity_id}, **data)

    async def get_state_raw(self, entity_id: str) -> "HassStateDict":
        """Get the state of a specific entity.

        Args:
            entity_id: The ID of the entity to get the state for.

        Returns:
            The state of the entity as raw data.
        """
        url = f"states/{entity_id}"
        response = await self.get_rest_request(url)
        return await response.json()

    async def entity_exists(self, entity_id: str) -> bool:
        """Check if a specific entity exists.

        Args:
            entity_id: The ID of the entity to check.

        Returns:
            True if the entity exists, False otherwise.
        """
        try:
            url = f"states/{entity_id}"
            response = await self.rest_request("GET", url, suppress_error_message=True)
            await response.json()
            return True
        except EntityNotFoundError:
            return False

    async def get_entity(self, entity_id: str, model: type["EntityT"]) -> "EntityT":
        """Get an entity object for a specific entity.

        Args:
            entity_id: The ID of the entity to get.
            model: The model class to use for the entity.

        Returns:
            The entity object.
        """
        if not issubclass(model, BaseEntity):  # runtime check
            raise TypeError(f"Model {model!r} is not a valid BaseEntity subclass")

        raw = await self.get_state_raw(entity_id)

        return model.model_validate({"state": raw})

    async def get_entity_or_none(self, entity_id: str, model: type["EntityT"]) -> "EntityT | None":
        """Get an entity object for a specific entity, or None if it does not exist.

        Args:
            entity_id: The ID of the entity to get.
            model: The model class to use for the entity.

        Returns:
            The entity object, or None if it does not exist.
        """
        try:
            return await self.get_entity(entity_id, model)
        except EntityNotFoundError:
            return None

    async def get_state(self, entity_id: str) -> "BaseState":
        """Get the state of a specific entity.

        Args:
            entity_id: The ID of the entity to get the state for.

        Returns:
            The state of the entity converted to the specified model type.
        """
        raw = await self.get_state_raw(entity_id)
        return self.hassette.state_registry.try_convert_state(raw, entity_id)

    async def get_state_or_none(self, entity_id: str) -> "BaseState | None":
        """Get the state of a specific entity, or None if it does not exist.

        Args:
            entity_id: The ID of the entity to get the state for.

        Returns:
            The state of the entity converted to the specified model type, or None if it does not exist.
        """
        try:
            return await self.get_state(entity_id)
        except EntityNotFoundError:
            return None

    async def get_state_value(self, entity_id: str) -> Any:
        """Get the state of a specific entity without converting it to a state object.

        Args:
            entity_id: The ID of the entity to get the state for.

        Returns:
            The state of the entity as raw data.

        Note:
            While most default methods in this library work with state objects for
            strong typing, this method is designed to return the raw state value,
            as it is likely overkill to convert it to a state object for simple state value retrieval.
        """
        entity = await self.get_state_raw(entity_id)
        state = entity.get("state")
        return state

    async def get_attribute(self, entity_id: str, attribute: str) -> Any | FalseySentinel:
        """Get a specific attribute of an entity.

        Args:
            entity_id: The ID of the entity to get the attribute for.
            attribute: The name of the attribute to retrieve. Can be a dot-separated path for nested attributes.

        Returns:
            The value of the specified attribute, or MISSING_VALUE sentinel if the attribute does not exist.
        """
        entity = await self.get_state(entity_id)
        return get_path(attribute)(entity.attributes)

    async def get_history(
        self,
        entity_id: str,
        start_time: PlainDateTime | ZonedDateTime | Date | str,
        end_time: PlainDateTime | ZonedDateTime | Date | str | None = None,
        significant_changes_only: bool = False,
        minimal_response: bool = False,
        no_attributes: bool = False,
    ) -> list[HistoryEntry]:
        """Get the history of a specific entity.

        Args:
            entity_id: The ID of the entity to get the history for.
            start_time: The start time for the history range.
            end_time: The end time for the history range.
            significant_changes_only: Whether to only include significant changes.
            minimal_response: Whether to request a minimal response.
            no_attributes: Whether to exclude attributes from the response.

        Returns:
            A list of history entries for the specified entity.
        """
        if "," in entity_id:
            raise ValueError("Entity ID should not contain commas. Use `get_histories` for multiple entities.")

        entries = await self._api_service.get_history_raw(
            entity_id=entity_id,
            start_time=start_time,
            end_time=end_time,
            significant_changes_only=significant_changes_only,
            minimal_response=minimal_response,
            no_attributes=no_attributes,
        )

        if not entries:
            return []

        if len(entries) != 1:
            raise RuntimeError(f"Expected a single list of history entries from HA, got {len(entries)}")

        converted = [HistoryEntry.model_validate(entry) for entry in entries[0]]

        return converted

    async def get_histories(
        self,
        entity_ids: list[str],
        start_time: PlainDateTime | ZonedDateTime | Date | str,
        end_time: PlainDateTime | ZonedDateTime | Date | str | None = None,
        significant_changes_only: bool = False,
        minimal_response: bool = False,
        no_attributes: bool = False,
    ) -> dict[str, list[HistoryEntry]]:
        """Get the history for multiple entities.

        Args:
            entity_ids: The IDs of the entities to get the history for.
            start_time: The start time for the history range.
            end_time: The end time for the history range.
            significant_changes_only: Whether to only include significant changes.
            minimal_response: Whether to request a minimal response.
            no_attributes: Whether to exclude attributes from the response.

        Returns:
            A dictionary mapping entity IDs to their respective history entries.
        """
        entity_id = ",".join(entity_ids)

        entries = await self._api_service.get_history_raw(
            entity_id=entity_id,
            start_time=start_time,
            end_time=end_time,
            significant_changes_only=significant_changes_only,
            minimal_response=minimal_response,
            no_attributes=no_attributes,
        )

        if not entries:
            return {}

        converted = {}
        for history_list in entries:
            converted[history_list[0]["entity_id"]] = [HistoryEntry.model_validate(entry) for entry in history_list]

        return converted

    async def get_logbook(
        self,
        entity_id: str,
        start_time: PlainDateTime | ZonedDateTime | Date | str,
        end_time: PlainDateTime | ZonedDateTime | Date | str,
    ) -> list[dict[str, Any]]:
        """Get the logbook entries for a specific entity.

        Args:
            entity_id: The ID of the entity to get the logbook entries for.
            start_time: The start time for the logbook range.
            end_time: The end time for the logbook range.

        Returns:
            A list of logbook entries for the specified entity.
        """
        url = f"logbook/{format_time_param(start_time)}"
        params = {"entity": entity_id, "end_time": end_time}

        response = await self.get_rest_request(url, params=params)

        return await response.json()

    def set_state(
        self,
        entity_id: str | StrEnum,
        state: Any,
        attributes: dict[str, Any] | None = None,
    ) -> "Coroutine[Any, Any, dict]":
        """Set the state of a specific entity.

        Must be awaited — a forgotten ``await`` is reported per ``forgotten_await_behavior`` (default: warn).

        Args:
            entity_id: The ID of the entity to set the state for.
            state: The new state value to set.
            attributes: Additional attributes to set for the entity.

        Returns:
            The response from Home Assistant after setting the state.
        """
        # Cheap path — see fire_event (same rationale for all api methods)
        source_location = capture_source_location()
        # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
        return guard_await(
            self._set_state(entity_id, state, attributes),
            owner=self.parent,
            source_location=source_location,
            method_name="set_state",
        )

    async def _set_state(
        self,
        entity_id: str | StrEnum,
        state: Any,
        attributes: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Async body for set_state."""
        entity_id = str(entity_id)

        attributes = attributes or {}
        curr_attributes = {}

        if await self.entity_exists(entity_id):
            curr_attributes = (await self.get_state_raw(entity_id)).get("attributes", {}) or {}

        # Merge current attributes with new attributes.
        # This client-side merge is load-bearing: HA's POST /api/states/{id} REPLACES the attribute
        # set rather than merging it (verified against a live instance 2026-06-21 — a partial POST
        # dropped the un-named attributes). Without this read-then-merge, every partial attribute
        # write would wipe the attributes it doesn't mention. See tests/system/test_api.py
        # ::test_set_state_preserves_unnamed_attributes.
        new_attributes = curr_attributes | attributes

        url = f"states/{entity_id}"
        data = {"state": state, "attributes": new_attributes}

        response = await self.post_rest_request(url, data=data)
        return await response.json()

    async def get_camera_image(
        self,
        entity_id: str,
        timestamp: PlainDateTime | ZonedDateTime | Date | str | None = None,
    ) -> bytes:
        """Get the latest camera image for a specific entity.

        Args:
            entity_id: The ID of the camera entity to get the image for.
            timestamp: The timestamp for the image. If None, the latest image is returned.

        Returns:
            The camera image data.
        """
        url = f"camera_proxy/{entity_id}"
        params = {}
        if timestamp:
            params["timestamp"] = timestamp

        response = await self.get_rest_request(url, params=params)

        return await response.read()

    async def get_calendars(self) -> list[dict[str, Any]]:
        """Get the list of calendars."""
        url = "calendars"
        response = await self.get_rest_request(url)
        return await response.json()

    async def get_calendar_events(
        self,
        calendar_id: str,
        start_time: PlainDateTime | ZonedDateTime | Date | str,
        end_time: PlainDateTime | ZonedDateTime | Date | str,
    ) -> list[dict[str, Any]]:
        """Get events from a specific calendar.

        Args:
            calendar_id: The ID of the calendar to get events from.
            start_time: The start time for the event range.
            end_time: The end time for the event range.

        Returns:
            A list of calendar events.
        """
        url = f"calendars/{calendar_id}/events"
        params = {"start": start_time, "end": end_time}

        response = await self.get_rest_request(url, params=params)
        return await response.json()

    async def render_template(
        self,
        template: str,
        variables: dict[str, Any] | None = None,
    ) -> str:
        """Render a template with given variables.

        Args:
            template: The template string to render.
            variables: Variables to use in the template.

        Returns:
            The rendered template result.
        """
        url = "template"
        data = {"template": template, "variables": variables or {}}

        response = await self.post_rest_request(url, data=data)
        return await response.text()

    async def delete_entity(self, entity_id: str) -> None:
        """Delete a specific entity.

        Args:
            entity_id: The ID of the entity to delete.

        Raises:
            RuntimeError: If the deletion fails.
        """
        url = f"states/{entity_id}"

        response = await self.rest_request("DELETE", url)

        if response.status != HTTPStatus.NO_CONTENT:
            raise RuntimeError(f"Failed to delete entity {entity_id}: {response.status} - {response.reason}")

sync: ApiSyncFacade = self.add_child(ApiSyncFacade, api=self) instance-attribute

Synchronous facade for the API service.

helpers: HelperClient = self.add_child(HelperClient, api=self) instance-attribute

Client for CRUD operations on Home Assistant helper entities.

config_log_level: LOG_LEVEL_TYPE property

Return the log level from the config for this resource.

ws_send_and_wait(**data: Any) -> Any async

Send a WebSocket message and wait for a response.

Source code in src/hassette/api/api.py

async def ws_send_and_wait(self, **data: Any) -> Any:
    """Send a WebSocket message and wait for a response."""
    return await self._api_service.ws_conn.send_and_wait(**data)

ws_send_json(**data: Any) -> None async

Send a WebSocket message without waiting for a response.

Source code in src/hassette/api/api.py

async def ws_send_json(self, **data: Any) -> None:
    """Send a WebSocket message without waiting for a response."""
    await self._api_service.ws_conn.send_json(**data)

rest_request(method: str, url: str, params: dict[str, Any] | None = None, data: dict[str, Any] | None = None, suppress_error_message: bool = False, **kwargs: Any) -> aiohttp.ClientResponse async

Make a REST request to the Home Assistant API.

Parameters:

Name	Type	Description	Default
method	str	The HTTP method to use (e.g., "GET", "POST").	required
url	str	The URL endpoint for the request.	required
params	dict[str, Any] | None	Query parameters for the request.	None
data	dict[str, Any] | None	JSON payload for the request.	None
suppress_error_message	bool	Whether to suppress error messages.	False

Returns:

Type	Description
ClientResponse	The response from the API.

Source code in src/hassette/api/api.py

async def rest_request(
    self,
    method: str,
    url: str,
    params: dict[str, Any] | None = None,
    data: dict[str, Any] | None = None,
    suppress_error_message: bool = False,
    **kwargs: Any,
) -> aiohttp.ClientResponse:
    """Make a REST request to the Home Assistant API.

    Args:
        method: The HTTP method to use (e.g., "GET", "POST").
        url: The URL endpoint for the request.
        params: Query parameters for the request.
        data: JSON payload for the request.
        suppress_error_message: Whether to suppress error messages.

    Returns:
        The response from the API.
    """
    return await self._api_service.rest_request(
        method, url, params=params, data=data, suppress_error_message=suppress_error_message, **kwargs
    )

get_rest_request(url: str, params: dict[str, Any] | None = None, **kwargs: Any) -> aiohttp.ClientResponse async

Make a GET request to the Home Assistant API.

Parameters:

Name	Type	Description	Default
url	str	The URL endpoint for the request.	required
params	dict[str, Any] | None	Query parameters for the request.	None
kwargs	Any	Additional keyword arguments to pass to the request.	{}

Returns:

Type	Description
ClientResponse	The response from the API.

Source code in src/hassette/api/api.py

async def get_rest_request(
    self, url: str, params: dict[str, Any] | None = None, **kwargs: Any
) -> aiohttp.ClientResponse:
    """Make a GET request to the Home Assistant API.

    Args:
        url: The URL endpoint for the request.
        params: Query parameters for the request.
        kwargs: Additional keyword arguments to pass to the request.

    Returns:
        The response from the API.
    """
    return await self.rest_request("GET", url, params=params, **kwargs)

post_rest_request(url: str, data: dict[str, Any] | None = None, **kwargs: Any) -> aiohttp.ClientResponse async

Make a POST request to the Home Assistant API.

Parameters:

Name	Type	Description	Default
url	str	The URL endpoint for the request.	required
data	dict[str, Any] | None	JSON payload for the request.	None
kwargs	Any	Additional keyword arguments to pass to the request.	{}

Returns:

Type	Description
ClientResponse	The response from the API.

Source code in src/hassette/api/api.py

async def post_rest_request(
    self, url: str, data: dict[str, Any] | None = None, **kwargs: Any
) -> aiohttp.ClientResponse:
    """Make a POST request to the Home Assistant API.

    Args:
        url: The URL endpoint for the request.
        data: JSON payload for the request.
        kwargs: Additional keyword arguments to pass to the request.

    Returns:
        The response from the API.
    """
    return await self.rest_request("POST", url, data=data, **kwargs)

delete_rest_request(url: str, **kwargs: Any) -> aiohttp.ClientResponse async

Make a DELETE request to the Home Assistant API.

Parameters:

Name	Type	Description	Default
url	str	The URL endpoint for the request.	required
kwargs	Any	Additional keyword arguments to pass to the request.	{}

Returns:

Type	Description
ClientResponse	The response from the API.

Source code in src/hassette/api/api.py

async def delete_rest_request(self, url: str, **kwargs: Any) -> aiohttp.ClientResponse:
    """Make a DELETE request to the Home Assistant API.

    Args:
        url: The URL endpoint for the request.
        kwargs: Additional keyword arguments to pass to the request.

    Returns:
        The response from the API.
    """
    return await self.rest_request("DELETE", url, **kwargs)

get_states_raw() -> list[HassStateDict] async

Get all entities in Home Assistant as raw dictionaries.

Returns:

Type	Description
list[HassStateDict]	A list of states as dictionaries.

Source code in src/hassette/api/api.py

async def get_states_raw(self) -> list["HassStateDict"]:
    """Get all entities in Home Assistant as raw dictionaries.

    Returns:
        A list of states as dictionaries.
    """
    val = await self.ws_send_and_wait(type="get_states")
    return _expect_list(val, "get_states")

get_states() -> list[BaseState] async

Get all entities in Home Assistant, converted to their appropriate state types.

If a state fails to convert, it is skipped with an error logged. If there is no registered state class for a domain, the generic BaseState is used.

Returns:

Type	Description
list[BaseState]	A list of states, converted to their appropriate state types.

Source code in src/hassette/api/api.py

async def get_states(self) -> list["BaseState"]:
    """Get all entities in Home Assistant, converted to their appropriate state types.

    If a state fails to convert, it is skipped with an error logged. If there is no registered
    state class for a domain, the generic BaseState is used.

    Returns:
        A list of states, converted to their appropriate state types.
    """
    val = await self.get_states_raw()

    self.logger.debug("Converting states to specific state types")
    converted: list[BaseState] = []

    for raw_state in val:
        # the conversion method will handle logging any conversion errors
        with suppress(UnableToConvertStateError):
            state = self.hassette.state_registry.try_convert_state(raw_state)
            converted.append(state)

    return converted

get_config() -> dict[str, Any] async

Get the Home Assistant configuration.

Returns:

Type	Description
dict[str, Any]	The configuration data.

Source code in src/hassette/api/api.py

async def get_config(self) -> dict[str, Any]:
    """Get the Home Assistant configuration.

    Returns:
        The configuration data.
    """
    val = await self.ws_send_and_wait(type="get_config")
    return _expect_dict(val, "get_config")

get_services() -> dict[str, Any] async

Get the available services in Home Assistant.

Returns:

Type	Description
dict[str, Any]	The services data.

Source code in src/hassette/api/api.py

async def get_services(self) -> dict[str, Any]:
    """Get the available services in Home Assistant.

    Returns:
        The services data.
    """
    val = await self.ws_send_and_wait(type="get_services")
    return _expect_dict(val, "get_services")

get_panels() -> dict[str, Any] async

Get the available panels in Home Assistant.

Returns:

Type	Description
dict[str, Any]	The panels data.

Source code in src/hassette/api/api.py

async def get_panels(self) -> dict[str, Any]:
    """Get the available panels in Home Assistant.

    Returns:
        The panels data.
    """
    val = await self.ws_send_and_wait(type="get_panels")
    return _expect_dict(val, "get_panels")

fire_event(event_type: str, event_data: dict[str, Any] | None = None) -> Coroutine[Any, Any, dict[str, Any]]

Fire a custom event in Home Assistant.

Must be awaited — a forgotten await is reported per forgotten_await_behavior (default: warn).

Parameters:

Name	Type	Description	Default
event_type	str	The type of the event to fire (e.g., "custom_event").	required
event_data	dict[str, Any] | None	Additional data to include with the event.	None

Returns:

Type	Description
Coroutine[Any, Any, dict[str, Any]]	The response from Home Assistant.

Source code in src/hassette/api/api.py

def fire_event(
    self,
    event_type: str,
    event_data: dict[str, Any] | None = None,
) -> "Coroutine[Any, Any, dict[str, Any]]":
    """Fire a custom event in Home Assistant.

    Must be awaited — a forgotten ``await`` is reported per ``forgotten_await_behavior`` (default: warn).

    Args:
        event_type: The type of the event to fire (e.g., "custom_event").
        event_data: Additional data to include with the event.

    Returns:
        The response from Home Assistant.
    """
    # Cheap path: no DB-record telemetry on api fire-and-forget methods
    # (unlike bus/scheduler listeners) — only warning attribution needs the location here.
    source_location = capture_source_location()
    # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
    return guard_await(
        self._fire_event(event_type, event_data),
        owner=self.parent,
        source_location=source_location,
        method_name="fire_event",
    )

call_service(domain: str, service: str, target: dict[str, str] | dict[str, list[str]] | None = None, return_response: bool | None = False, **data: Any) -> Coroutine[Any, Any, ServiceResponse | None]

call_service(
    domain: str,
    service: str,
    target: dict[str, str]
    | dict[str, list[str]]
    | None = None,
    return_response: typing.Literal[False] | None = None,
    **data: Any,
) -> Coroutine[Any, Any, None]

call_service(
    domain: str,
    service: str,
    target: dict[str, str]
    | dict[str, list[str]]
    | None = None,
    return_response: Literal[True] = True,
    **data: Any,
) -> Coroutine[Any, Any, ServiceResponse]

Call a Home Assistant service.

Must be awaited — a forgotten await is reported per forgotten_await_behavior (default: warn).

Parameters:

Name	Type	Description	Default
domain	str	The domain of the service (e.g., "light").	required
service	str	The name of the service to call (e.g., "turn_on").	required
target	dict[str, str] | dict[str, list[str]] | None	Target entity IDs or areas.	None
return_response	bool | None	Whether to return the response from Home Assistant. Defaults to False.	False
**data	Any	Additional data to send with the service call.	{}

Returns:

Type	Description
Coroutine[Any, Any, ServiceResponse | None]	ServiceResponse | None: The response from Home Assistant if return_response is True. Otherwise None.

Source code in src/hassette/api/api.py

def call_service(
    self,
    domain: str,
    service: str,
    target: dict[str, str] | dict[str, list[str]] | None = None,
    return_response: bool | None = False,
    **data: Any,
) -> "Coroutine[Any, Any, ServiceResponse | None]":
    """Call a Home Assistant service.

    Must be awaited — a forgotten ``await`` is reported per ``forgotten_await_behavior`` (default: warn).

    Args:
        domain: The domain of the service (e.g., "light").
        service: The name of the service to call (e.g., "turn_on").
        target: Target entity IDs or areas.
        return_response: Whether to return the response from Home Assistant. Defaults to False.
        **data: Additional data to send with the service call.

    Returns:
        ServiceResponse | None: The response from Home Assistant if return_response is True. Otherwise None.
    """
    # Cheap path — see fire_event (same rationale for all api methods)
    source_location = capture_source_location()
    # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
    return guard_await(
        self._call_service(domain, service, target, return_response, **data),
        owner=self.parent,
        source_location=source_location,
        method_name="call_service",
    )

turn_on(entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> Coroutine[Any, Any, None]

Turn on a specific entity in Home Assistant.

Must be awaited — a forgotten await is reported per forgotten_await_behavior (default: warn).

Parameters:

Name	Type	Description	Default
entity_id	str | StrEnum	The ID of the entity to turn on (e.g., "light.office").	required
domain	str | None	The domain to use for the service call. Defaults to the entity's domain, derived from entity_id (e.g., "light.office" derives "light"). Pass explicitly to override — for example, domain="homeassistant" to use the generic (deprecated) service.	None
**data	Any	Additional service data forwarded to the service call.	{}

Source code in src/hassette/api/api.py

def turn_on(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> "Coroutine[Any, Any, None]":
    """Turn on a specific entity in Home Assistant.

    Must be awaited — a forgotten ``await`` is reported per ``forgotten_await_behavior`` (default: warn).

    Args:
        entity_id: The ID of the entity to turn on (e.g., "light.office").
        domain: The domain to use for the service call. Defaults to the entity's domain,
            derived from ``entity_id`` (e.g., ``"light.office"`` derives ``"light"``). Pass
            explicitly to override — for example, ``domain="homeassistant"`` to use the
            generic (deprecated) service.
        **data: Additional service data forwarded to the service call.

    """
    entity_id = str(entity_id)
    if domain is None:
        domain = entity_id.split(".", 1)[0]
    return self.call_service(domain=domain, service="turn_on", target={"entity_id": entity_id}, **data)

turn_off(entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> Coroutine[Any, Any, None]

Turn off a specific entity in Home Assistant.

Must be awaited — a forgotten await is reported per forgotten_await_behavior (default: warn).

Parameters:

Name	Type	Description	Default
entity_id	str | StrEnum	The ID of the entity to turn off (e.g., "light.office").	required
domain	str | None	The domain to use for the service call. Defaults to the entity's domain, derived from entity_id (e.g., "light.office" derives "light"). Pass explicitly to override — for example, domain="homeassistant" to use the generic (deprecated) service.	None
**data	Any	Additional service data forwarded to the service call.	{}

Source code in src/hassette/api/api.py

def turn_off(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> "Coroutine[Any, Any, None]":
    """Turn off a specific entity in Home Assistant.

    Must be awaited — a forgotten ``await`` is reported per ``forgotten_await_behavior`` (default: warn).

    Args:
        entity_id: The ID of the entity to turn off (e.g., "light.office").
        domain: The domain to use for the service call. Defaults to the entity's domain,
            derived from ``entity_id`` (e.g., ``"light.office"`` derives ``"light"``). Pass
            explicitly to override — for example, ``domain="homeassistant"`` to use the
            generic (deprecated) service.
        **data: Additional service data forwarded to the service call.

    """
    entity_id = str(entity_id)
    if domain is None:
        domain = entity_id.split(".", 1)[0]
    return self.call_service(domain=domain, service="turn_off", target={"entity_id": entity_id}, **data)

toggle(entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> Coroutine[Any, Any, None]

Toggle a specific entity in Home Assistant.

Must be awaited — a forgotten await is reported per forgotten_await_behavior (default: warn).

Parameters:

Name	Type	Description	Default
entity_id	str | StrEnum	The ID of the entity to toggle (e.g., "light.office").	required
domain	str | None	The domain to use for the service call. Defaults to the entity's domain, derived from entity_id (e.g., "light.office" derives "light"). Pass explicitly to override — for example, domain="homeassistant" to use the generic (deprecated) service.	None
**data	Any	Additional service data forwarded to the service call.	{}

Source code in src/hassette/api/api.py

def toggle(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> "Coroutine[Any, Any, None]":
    """Toggle a specific entity in Home Assistant.

    Must be awaited — a forgotten ``await`` is reported per ``forgotten_await_behavior`` (default: warn).

    Args:
        entity_id: The ID of the entity to toggle (e.g., "light.office").
        domain: The domain to use for the service call. Defaults to the entity's domain,
            derived from ``entity_id`` (e.g., ``"light.office"`` derives ``"light"``). Pass
            explicitly to override — for example, ``domain="homeassistant"`` to use the
            generic (deprecated) service.
        **data: Additional service data forwarded to the service call.

    """
    entity_id = str(entity_id)
    if domain is None:
        domain = entity_id.split(".", 1)[0]
    return self.call_service(domain=domain, service="toggle", target={"entity_id": entity_id}, **data)

get_state_raw(entity_id: str) -> HassStateDict async

Get the state of a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get the state for.	required

Returns:

Type	Description
HassStateDict	The state of the entity as raw data.

Source code in src/hassette/api/api.py

async def get_state_raw(self, entity_id: str) -> "HassStateDict":
    """Get the state of a specific entity.

    Args:
        entity_id: The ID of the entity to get the state for.

    Returns:
        The state of the entity as raw data.
    """
    url = f"states/{entity_id}"
    response = await self.get_rest_request(url)
    return await response.json()

entity_exists(entity_id: str) -> bool async

Check if a specific entity exists.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to check.	required

Returns:

Type	Description
bool	True if the entity exists, False otherwise.

Source code in src/hassette/api/api.py

async def entity_exists(self, entity_id: str) -> bool:
    """Check if a specific entity exists.

    Args:
        entity_id: The ID of the entity to check.

    Returns:
        True if the entity exists, False otherwise.
    """
    try:
        url = f"states/{entity_id}"
        response = await self.rest_request("GET", url, suppress_error_message=True)
        await response.json()
        return True
    except EntityNotFoundError:
        return False

get_entity(entity_id: str, model: type[EntityT]) -> EntityT async

Get an entity object for a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get.	required
model	type[EntityT]	The model class to use for the entity.	required

Returns:

Type	Description
EntityT	The entity object.

Source code in src/hassette/api/api.py

async def get_entity(self, entity_id: str, model: type["EntityT"]) -> "EntityT":
    """Get an entity object for a specific entity.

    Args:
        entity_id: The ID of the entity to get.
        model: The model class to use for the entity.

    Returns:
        The entity object.
    """
    if not issubclass(model, BaseEntity):  # runtime check
        raise TypeError(f"Model {model!r} is not a valid BaseEntity subclass")

    raw = await self.get_state_raw(entity_id)

    return model.model_validate({"state": raw})

get_entity_or_none(entity_id: str, model: type[EntityT]) -> EntityT | None async

Get an entity object for a specific entity, or None if it does not exist.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get.	required
model	type[EntityT]	The model class to use for the entity.	required

Returns:

Type	Description
EntityT | None	The entity object, or None if it does not exist.

Source code in src/hassette/api/api.py

async def get_entity_or_none(self, entity_id: str, model: type["EntityT"]) -> "EntityT | None":
    """Get an entity object for a specific entity, or None if it does not exist.

    Args:
        entity_id: The ID of the entity to get.
        model: The model class to use for the entity.

    Returns:
        The entity object, or None if it does not exist.
    """
    try:
        return await self.get_entity(entity_id, model)
    except EntityNotFoundError:
        return None

get_state(entity_id: str) -> BaseState async

Get the state of a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get the state for.	required

Returns:

Type	Description
BaseState	The state of the entity converted to the specified model type.

Source code in src/hassette/api/api.py

async def get_state(self, entity_id: str) -> "BaseState":
    """Get the state of a specific entity.

    Args:
        entity_id: The ID of the entity to get the state for.

    Returns:
        The state of the entity converted to the specified model type.
    """
    raw = await self.get_state_raw(entity_id)
    return self.hassette.state_registry.try_convert_state(raw, entity_id)

get_state_or_none(entity_id: str) -> BaseState | None async

Get the state of a specific entity, or None if it does not exist.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get the state for.	required

Returns:

Type	Description
BaseState | None	The state of the entity converted to the specified model type, or None if it does not exist.

Source code in src/hassette/api/api.py

async def get_state_or_none(self, entity_id: str) -> "BaseState | None":
    """Get the state of a specific entity, or None if it does not exist.

    Args:
        entity_id: The ID of the entity to get the state for.

    Returns:
        The state of the entity converted to the specified model type, or None if it does not exist.
    """
    try:
        return await self.get_state(entity_id)
    except EntityNotFoundError:
        return None

get_state_value(entity_id: str) -> Any async

Get the state of a specific entity without converting it to a state object.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get the state for.	required

Returns:

Type	Description
Any	The state of the entity as raw data.

Note

While most default methods in this library work with state objects for strong typing, this method is designed to return the raw state value, as it is likely overkill to convert it to a state object for simple state value retrieval.

Source code in src/hassette/api/api.py

async def get_state_value(self, entity_id: str) -> Any:
    """Get the state of a specific entity without converting it to a state object.

    Args:
        entity_id: The ID of the entity to get the state for.

    Returns:
        The state of the entity as raw data.

    Note:
        While most default methods in this library work with state objects for
        strong typing, this method is designed to return the raw state value,
        as it is likely overkill to convert it to a state object for simple state value retrieval.
    """
    entity = await self.get_state_raw(entity_id)
    state = entity.get("state")
    return state

get_attribute(entity_id: str, attribute: str) -> Any | FalseySentinel async

Get a specific attribute of an entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get the attribute for.	required
attribute	str	The name of the attribute to retrieve. Can be a dot-separated path for nested attributes.	required

Returns:

Type	Description
Any | FalseySentinel	The value of the specified attribute, or MISSING_VALUE sentinel if the attribute does not exist.

Source code in src/hassette/api/api.py

async def get_attribute(self, entity_id: str, attribute: str) -> Any | FalseySentinel:
    """Get a specific attribute of an entity.

    Args:
        entity_id: The ID of the entity to get the attribute for.
        attribute: The name of the attribute to retrieve. Can be a dot-separated path for nested attributes.

    Returns:
        The value of the specified attribute, or MISSING_VALUE sentinel if the attribute does not exist.
    """
    entity = await self.get_state(entity_id)
    return get_path(attribute)(entity.attributes)

get_history(entity_id: str, start_time: PlainDateTime | ZonedDateTime | Date | str, end_time: PlainDateTime | ZonedDateTime | Date | str | None = None, significant_changes_only: bool = False, minimal_response: bool = False, no_attributes: bool = False) -> list[HistoryEntry] async

Get the history of a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get the history for.	required
start_time	PlainDateTime | ZonedDateTime | Date | str	The start time for the history range.	required
end_time	PlainDateTime | ZonedDateTime | Date | str | None	The end time for the history range.	None
significant_changes_only	bool	Whether to only include significant changes.	False
minimal_response	bool	Whether to request a minimal response.	False
no_attributes	bool	Whether to exclude attributes from the response.	False

Returns:

Type	Description
list[HistoryEntry]	A list of history entries for the specified entity.

Source code in src/hassette/api/api.py

async def get_history(
    self,
    entity_id: str,
    start_time: PlainDateTime | ZonedDateTime | Date | str,
    end_time: PlainDateTime | ZonedDateTime | Date | str | None = None,
    significant_changes_only: bool = False,
    minimal_response: bool = False,
    no_attributes: bool = False,
) -> list[HistoryEntry]:
    """Get the history of a specific entity.

    Args:
        entity_id: The ID of the entity to get the history for.
        start_time: The start time for the history range.
        end_time: The end time for the history range.
        significant_changes_only: Whether to only include significant changes.
        minimal_response: Whether to request a minimal response.
        no_attributes: Whether to exclude attributes from the response.

    Returns:
        A list of history entries for the specified entity.
    """
    if "," in entity_id:
        raise ValueError("Entity ID should not contain commas. Use `get_histories` for multiple entities.")

    entries = await self._api_service.get_history_raw(
        entity_id=entity_id,
        start_time=start_time,
        end_time=end_time,
        significant_changes_only=significant_changes_only,
        minimal_response=minimal_response,
        no_attributes=no_attributes,
    )

    if not entries:
        return []

    if len(entries) != 1:
        raise RuntimeError(f"Expected a single list of history entries from HA, got {len(entries)}")

    converted = [HistoryEntry.model_validate(entry) for entry in entries[0]]

    return converted

get_histories(entity_ids: list[str], start_time: PlainDateTime | ZonedDateTime | Date | str, end_time: PlainDateTime | ZonedDateTime | Date | str | None = None, significant_changes_only: bool = False, minimal_response: bool = False, no_attributes: bool = False) -> dict[str, list[HistoryEntry]] async

Get the history for multiple entities.

Parameters:

Name	Type	Description	Default
entity_ids	list[str]	The IDs of the entities to get the history for.	required
start_time	PlainDateTime | ZonedDateTime | Date | str	The start time for the history range.	required
end_time	PlainDateTime | ZonedDateTime | Date | str | None	The end time for the history range.	None
significant_changes_only	bool	Whether to only include significant changes.	False
minimal_response	bool	Whether to request a minimal response.	False
no_attributes	bool	Whether to exclude attributes from the response.	False

Returns:

Type	Description
dict[str, list[HistoryEntry]]	A dictionary mapping entity IDs to their respective history entries.

Source code in src/hassette/api/api.py

async def get_histories(
    self,
    entity_ids: list[str],
    start_time: PlainDateTime | ZonedDateTime | Date | str,
    end_time: PlainDateTime | ZonedDateTime | Date | str | None = None,
    significant_changes_only: bool = False,
    minimal_response: bool = False,
    no_attributes: bool = False,
) -> dict[str, list[HistoryEntry]]:
    """Get the history for multiple entities.

    Args:
        entity_ids: The IDs of the entities to get the history for.
        start_time: The start time for the history range.
        end_time: The end time for the history range.
        significant_changes_only: Whether to only include significant changes.
        minimal_response: Whether to request a minimal response.
        no_attributes: Whether to exclude attributes from the response.

    Returns:
        A dictionary mapping entity IDs to their respective history entries.
    """
    entity_id = ",".join(entity_ids)

    entries = await self._api_service.get_history_raw(
        entity_id=entity_id,
        start_time=start_time,
        end_time=end_time,
        significant_changes_only=significant_changes_only,
        minimal_response=minimal_response,
        no_attributes=no_attributes,
    )

    if not entries:
        return {}

    converted = {}
    for history_list in entries:
        converted[history_list[0]["entity_id"]] = [HistoryEntry.model_validate(entry) for entry in history_list]

    return converted

get_logbook(entity_id: str, start_time: PlainDateTime | ZonedDateTime | Date | str, end_time: PlainDateTime | ZonedDateTime | Date | str) -> list[dict[str, Any]] async

Get the logbook entries for a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get the logbook entries for.	required
start_time	PlainDateTime | ZonedDateTime | Date | str	The start time for the logbook range.	required
end_time	PlainDateTime | ZonedDateTime | Date | str	The end time for the logbook range.	required

Returns:

Type	Description
list[dict[str, Any]]	A list of logbook entries for the specified entity.

Source code in src/hassette/api/api.py

async def get_logbook(
    self,
    entity_id: str,
    start_time: PlainDateTime | ZonedDateTime | Date | str,
    end_time: PlainDateTime | ZonedDateTime | Date | str,
) -> list[dict[str, Any]]:
    """Get the logbook entries for a specific entity.

    Args:
        entity_id: The ID of the entity to get the logbook entries for.
        start_time: The start time for the logbook range.
        end_time: The end time for the logbook range.

    Returns:
        A list of logbook entries for the specified entity.
    """
    url = f"logbook/{format_time_param(start_time)}"
    params = {"entity": entity_id, "end_time": end_time}

    response = await self.get_rest_request(url, params=params)

    return await response.json()

set_state(entity_id: str | StrEnum, state: Any, attributes: dict[str, Any] | None = None) -> Coroutine[Any, Any, dict]

Set the state of a specific entity.

Must be awaited — a forgotten await is reported per forgotten_await_behavior (default: warn).

Parameters:

Name	Type	Description	Default
entity_id	str | StrEnum	The ID of the entity to set the state for.	required
state	Any	The new state value to set.	required
attributes	dict[str, Any] | None	Additional attributes to set for the entity.	None

Returns:

Type	Description
Coroutine[Any, Any, dict]	The response from Home Assistant after setting the state.

Source code in src/hassette/api/api.py

def set_state(
    self,
    entity_id: str | StrEnum,
    state: Any,
    attributes: dict[str, Any] | None = None,
) -> "Coroutine[Any, Any, dict]":
    """Set the state of a specific entity.

    Must be awaited — a forgotten ``await`` is reported per ``forgotten_await_behavior`` (default: warn).

    Args:
        entity_id: The ID of the entity to set the state for.
        state: The new state value to set.
        attributes: Additional attributes to set for the entity.

    Returns:
        The response from Home Assistant after setting the state.
    """
    # Cheap path — see fire_event (same rationale for all api methods)
    source_location = capture_source_location()
    # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
    return guard_await(
        self._set_state(entity_id, state, attributes),
        owner=self.parent,
        source_location=source_location,
        method_name="set_state",
    )

get_camera_image(entity_id: str, timestamp: PlainDateTime | ZonedDateTime | Date | str | None = None) -> bytes async

Get the latest camera image for a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the camera entity to get the image for.	required
timestamp	PlainDateTime | ZonedDateTime | Date | str | None	The timestamp for the image. If None, the latest image is returned.	None

Returns:

Type	Description
bytes	The camera image data.

Source code in src/hassette/api/api.py

async def get_camera_image(
    self,
    entity_id: str,
    timestamp: PlainDateTime | ZonedDateTime | Date | str | None = None,
) -> bytes:
    """Get the latest camera image for a specific entity.

    Args:
        entity_id: The ID of the camera entity to get the image for.
        timestamp: The timestamp for the image. If None, the latest image is returned.

    Returns:
        The camera image data.
    """
    url = f"camera_proxy/{entity_id}"
    params = {}
    if timestamp:
        params["timestamp"] = timestamp

    response = await self.get_rest_request(url, params=params)

    return await response.read()

get_calendars() -> list[dict[str, Any]] async

Get the list of calendars.

Source code in src/hassette/api/api.py

async def get_calendars(self) -> list[dict[str, Any]]:
    """Get the list of calendars."""
    url = "calendars"
    response = await self.get_rest_request(url)
    return await response.json()

get_calendar_events(calendar_id: str, start_time: PlainDateTime | ZonedDateTime | Date | str, end_time: PlainDateTime | ZonedDateTime | Date | str) -> list[dict[str, Any]] async

Get events from a specific calendar.

Parameters:

Name	Type	Description	Default
calendar_id	str	The ID of the calendar to get events from.	required
start_time	PlainDateTime | ZonedDateTime | Date | str	The start time for the event range.	required
end_time	PlainDateTime | ZonedDateTime | Date | str	The end time for the event range.	required

Returns:

Type	Description
list[dict[str, Any]]	A list of calendar events.

Source code in src/hassette/api/api.py

async def get_calendar_events(
    self,
    calendar_id: str,
    start_time: PlainDateTime | ZonedDateTime | Date | str,
    end_time: PlainDateTime | ZonedDateTime | Date | str,
) -> list[dict[str, Any]]:
    """Get events from a specific calendar.

    Args:
        calendar_id: The ID of the calendar to get events from.
        start_time: The start time for the event range.
        end_time: The end time for the event range.

    Returns:
        A list of calendar events.
    """
    url = f"calendars/{calendar_id}/events"
    params = {"start": start_time, "end": end_time}

    response = await self.get_rest_request(url, params=params)
    return await response.json()

render_template(template: str, variables: dict[str, Any] | None = None) -> str async

Render a template with given variables.

Parameters:

Name	Type	Description	Default
template	str	The template string to render.	required
variables	dict[str, Any] | None	Variables to use in the template.	None

Returns:

Type	Description
str	The rendered template result.

Source code in src/hassette/api/api.py

async def render_template(
    self,
    template: str,
    variables: dict[str, Any] | None = None,
) -> str:
    """Render a template with given variables.

    Args:
        template: The template string to render.
        variables: Variables to use in the template.

    Returns:
        The rendered template result.
    """
    url = "template"
    data = {"template": template, "variables": variables or {}}

    response = await self.post_rest_request(url, data=data)
    return await response.text()

delete_entity(entity_id: str) -> None async

Delete a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to delete.	required

Raises:

Type	Description
RuntimeError	If the deletion fails.

Source code in src/hassette/api/api.py

async def delete_entity(self, entity_id: str) -> None:
    """Delete a specific entity.

    Args:
        entity_id: The ID of the entity to delete.

    Raises:
        RuntimeError: If the deletion fails.
    """
    url = f"states/{entity_id}"

    response = await self.rest_request("DELETE", url)

    if response.status != HTTPStatus.NO_CONTENT:
        raise RuntimeError(f"Failed to delete entity {entity_id}: {response.status} - {response.reason}")

HelperClient

Bases: Resource

Client for CRUD operations on Home Assistant helper entities.

Exposes 4 generic methods (list, create, update, delete) covering all 8 helper domains (input_boolean, input_number, input_text, input_select, input_datetime, input_button, counter, timer), plus 3 counter shortcuts (increment, decrement, reset). Each CRUD method dispatches to the correct HA WebSocket command and record type via a lookup table, with hand-maintained @overload declarations providing full static typing.

Source code in src/hassette/api/helpers.py

class HelperClient(Resource):
    """Client for CRUD operations on Home Assistant helper entities.

    Exposes 4 generic methods (`list`, `create`, `update`, `delete`) covering all 8 helper domains
    (`input_boolean`, `input_number`, `input_text`, `input_select`, `input_datetime`,
    `input_button`, `counter`, `timer`), plus 3 counter shortcuts (`increment`, `decrement`,
    `reset`). Each CRUD method dispatches to the correct HA WebSocket command and record type via
    a lookup table, with hand-maintained `@overload` declarations providing full static typing.
    """

    _api: "Api"

    def __init__(self, hassette: "Hassette", *, api: "Api", parent: Resource | None = None) -> None:
        super().__init__(hassette, parent=parent)
        self._api = api

    async def on_initialize(self) -> None:
        mark_ready(self, reason="Helper client initialized")

    @property
    def config_log_level(self) -> LOG_LEVEL_TYPE:
        """Return the log level from the config for this resource."""
        return self.hassette.config.logging.api

    # builtins.list: the method name "list" shadows the builtin in the class body,
    # so bare list[...] in annotations resolves to the method on Python < 3.14.

    @overload
    async def list(self, domain: Literal["input_boolean"]) -> builtins.list[InputBooleanRecord]: ...
    @overload
    async def list(self, domain: Literal["input_number"]) -> builtins.list[InputNumberRecord]: ...
    @overload
    async def list(self, domain: Literal["input_text"]) -> builtins.list[InputTextRecord]: ...
    @overload
    async def list(self, domain: Literal["input_select"]) -> builtins.list[InputSelectRecord]: ...
    @overload
    async def list(self, domain: Literal["input_datetime"]) -> builtins.list[InputDatetimeRecord]: ...
    @overload
    async def list(self, domain: Literal["input_button"]) -> builtins.list[InputButtonRecord]: ...
    @overload
    async def list(self, domain: Literal["counter"]) -> builtins.list[CounterRecord]: ...
    @overload
    async def list(self, domain: Literal["timer"]) -> builtins.list[TimerRecord]: ...

    async def list(self, domain: HelperDomain) -> builtins.list[Any]:
        """List all stored helpers for a given domain.

        Args:
            domain: The helper domain to list (e.g. "input_boolean", "counter").

        Returns:
            List of domain-specific Record instances representing stored configs.
        """
        record_type = DOMAIN_DISPATCH[domain]
        val = await _ws_helper_call(self._api, domain, "list")
        items = _expect_list(val, f"{domain}/list")
        self.logger.debug("Listed %d %s helpers", len(items), domain)
        return [record_type.model_validate(item) for item in items]

    # create dispatches on type(params)

    @overload
    async def create(self, params: CreateInputBooleanParams) -> InputBooleanRecord: ...
    @overload
    async def create(self, params: CreateInputNumberParams) -> InputNumberRecord: ...
    @overload
    async def create(self, params: CreateInputTextParams) -> InputTextRecord: ...
    @overload
    async def create(self, params: CreateInputSelectParams) -> InputSelectRecord: ...
    @overload
    async def create(self, params: CreateInputDatetimeParams) -> InputDatetimeRecord: ...
    @overload
    async def create(self, params: CreateInputButtonParams) -> InputButtonRecord: ...
    @overload
    async def create(self, params: CreateCounterParams) -> CounterRecord: ...
    @overload
    async def create(self, params: CreateTimerParams) -> TimerRecord: ...

    async def create(self, params: BaseModel) -> BaseModel:
        """Create a new helper.

        Args:
            params: Parameters for the new helper. The concrete type determines the domain
                and the return type via overload resolution.

        Returns:
            The stored record returned by Home Assistant.
        """
        domain, record_type = CREATE_DISPATCH[type(params)]
        val = await _ws_helper_call(self._api, domain, "create", **params.model_dump(exclude_unset=True))
        record = record_type.model_validate(_expect_dict(val, f"{domain}/create"))
        self.logger.info("Created %s helper %r", domain, record.id)  # pyright: ignore[reportAttributeAccessIssue]
        return record

    # update dispatches on type(params)

    @overload
    async def update(self, helper_id: str, params: UpdateInputBooleanParams) -> InputBooleanRecord: ...
    @overload
    async def update(self, helper_id: str, params: UpdateInputNumberParams) -> InputNumberRecord: ...
    @overload
    async def update(self, helper_id: str, params: UpdateInputTextParams) -> InputTextRecord: ...
    @overload
    async def update(self, helper_id: str, params: UpdateInputSelectParams) -> InputSelectRecord: ...
    @overload
    async def update(self, helper_id: str, params: UpdateInputDatetimeParams) -> InputDatetimeRecord: ...
    @overload
    async def update(self, helper_id: str, params: UpdateInputButtonParams) -> InputButtonRecord: ...
    @overload
    async def update(self, helper_id: str, params: UpdateCounterParams) -> CounterRecord: ...
    @overload
    async def update(self, helper_id: str, params: UpdateTimerParams) -> TimerRecord: ...

    async def update(self, helper_id: str, params: BaseModel) -> BaseModel:
        """Update an existing helper.

        Args:
            helper_id: The ID of the helper to update.
            params: Fields to update (unset fields are left unchanged). The concrete type
                determines the domain and the return type via overload resolution.

        Returns:
            The updated stored record.
        """
        domain, record_type, id_key = UPDATE_DISPATCH[type(params)]
        val = await _ws_helper_call(
            self._api, domain, "update", **{id_key: helper_id}, **params.model_dump(exclude_unset=True)
        )
        record = record_type.model_validate(_expect_dict(val, f"{domain}/update"))
        self.logger.debug("Updated %s helper %r", domain, helper_id)
        return record

    # delete dispatches on a Literal domain string

    @overload
    async def delete(self, domain: Literal["input_boolean"], helper_id: str) -> None: ...
    @overload
    async def delete(self, domain: Literal["input_number"], helper_id: str) -> None: ...
    @overload
    async def delete(self, domain: Literal["input_text"], helper_id: str) -> None: ...
    @overload
    async def delete(self, domain: Literal["input_select"], helper_id: str) -> None: ...
    @overload
    async def delete(self, domain: Literal["input_datetime"], helper_id: str) -> None: ...
    @overload
    async def delete(self, domain: Literal["input_button"], helper_id: str) -> None: ...
    @overload
    async def delete(self, domain: Literal["counter"], helper_id: str) -> None: ...
    @overload
    async def delete(self, domain: Literal["timer"], helper_id: str) -> None: ...

    async def delete(self, domain: HelperDomain, helper_id: str) -> None:
        """Delete a helper.

        Args:
            domain: The helper domain (e.g. "input_boolean", "counter").
            helper_id: The ID of the helper to delete.
        """
        id_key = ID_KEYS[domain]
        await _ws_helper_call(self._api, domain, "delete", **{id_key: helper_id})
        self.logger.debug("Deleted %s helper %r", domain, helper_id)

    # counter shortcuts
    # Counter service-call shortcuts (operate on live entity state, not stored
    # config). Use update() to change the stored initial/minimum/maximum.
    #
    # timer.start / timer.pause / timer.cancel are deliberately excluded: timer
    # service actions are one-off calls that benefit from the full call_service()
    # signature. Counter actions get wrappers because the pattern "increment on
    # every event" is common enough to warrant a two-word call site.

    async def increment(self, entity_id: str) -> None:
        """Increment a counter entity's current value (live state, not stored config).

        Args:
            entity_id: The entity ID of the counter (e.g. ``"counter.motion_count"``).
        """
        await self._api.call_service(
            "counter",
            "increment",
            target={"entity_id": entity_id},
            return_response=True,  # surfaces HA errors instead of fire-and-forget
        )
        self.logger.debug("Incremented counter %r", entity_id)

    async def decrement(self, entity_id: str) -> None:
        """Decrement a counter entity's current value (live state, not stored config).

        Args:
            entity_id: The entity ID of the counter (e.g. ``"counter.motion_count"``).
        """
        await self._api.call_service(
            "counter",
            "decrement",
            target={"entity_id": entity_id},
            return_response=True,
        )
        self.logger.debug("Decremented counter %r", entity_id)

    async def reset(self, entity_id: str) -> None:
        """Reset a counter entity's value to its configured initial (live state, not stored config).

        Args:
            entity_id: The entity ID of the counter (e.g. ``"counter.motion_count"``).
        """
        await self._api.call_service(
            "counter",
            "reset",
            target={"entity_id": entity_id},
            return_response=True,
        )
        self.logger.debug("Reset counter %r", entity_id)

config_log_level: LOG_LEVEL_TYPE property

Return the log level from the config for this resource.

list(domain: HelperDomain) -> builtins.list[Any] async

list(
    domain: Literal["input_boolean"],
) -> builtins.list[InputBooleanRecord]

list(
    domain: Literal["input_number"],
) -> builtins.list[InputNumberRecord]

list(
    domain: Literal["input_text"],
) -> builtins.list[InputTextRecord]

list(
    domain: Literal["input_select"],
) -> builtins.list[InputSelectRecord]

list(
    domain: Literal["input_datetime"],
) -> builtins.list[InputDatetimeRecord]

list(
    domain: Literal["input_button"],
) -> builtins.list[InputButtonRecord]

list(
    domain: Literal["counter"],
) -> builtins.list[CounterRecord]

list(
    domain: Literal["timer"],
) -> builtins.list[TimerRecord]

List all stored helpers for a given domain.

Parameters:

Name	Type	Description	Default
domain	HelperDomain	The helper domain to list (e.g. "input_boolean", "counter").	required

Returns:

Type	Description
list[Any]	List of domain-specific Record instances representing stored configs.

Source code in src/hassette/api/helpers.py

async def list(self, domain: HelperDomain) -> builtins.list[Any]:
    """List all stored helpers for a given domain.

    Args:
        domain: The helper domain to list (e.g. "input_boolean", "counter").

    Returns:
        List of domain-specific Record instances representing stored configs.
    """
    record_type = DOMAIN_DISPATCH[domain]
    val = await _ws_helper_call(self._api, domain, "list")
    items = _expect_list(val, f"{domain}/list")
    self.logger.debug("Listed %d %s helpers", len(items), domain)
    return [record_type.model_validate(item) for item in items]

create(params: BaseModel) -> BaseModel async

create(
    params: CreateInputBooleanParams,
) -> InputBooleanRecord

create(
    params: CreateInputNumberParams,
) -> InputNumberRecord

create(params: CreateInputTextParams) -> InputTextRecord

create(
    params: CreateInputSelectParams,
) -> InputSelectRecord

create(
    params: CreateInputDatetimeParams,
) -> InputDatetimeRecord

create(
    params: CreateInputButtonParams,
) -> InputButtonRecord

create(params: CreateCounterParams) -> CounterRecord

create(params: CreateTimerParams) -> TimerRecord

Create a new helper.

Parameters:

Name	Type	Description	Default
params	BaseModel	Parameters for the new helper. The concrete type determines the domain and the return type via overload resolution.	required

Returns:

Type	Description
BaseModel	The stored record returned by Home Assistant.

Source code in src/hassette/api/helpers.py

async def create(self, params: BaseModel) -> BaseModel:
    """Create a new helper.

    Args:
        params: Parameters for the new helper. The concrete type determines the domain
            and the return type via overload resolution.

    Returns:
        The stored record returned by Home Assistant.
    """
    domain, record_type = CREATE_DISPATCH[type(params)]
    val = await _ws_helper_call(self._api, domain, "create", **params.model_dump(exclude_unset=True))
    record = record_type.model_validate(_expect_dict(val, f"{domain}/create"))
    self.logger.info("Created %s helper %r", domain, record.id)  # pyright: ignore[reportAttributeAccessIssue]
    return record

update(helper_id: str, params: BaseModel) -> BaseModel async

update(
    helper_id: str, params: UpdateInputBooleanParams
) -> InputBooleanRecord

update(
    helper_id: str, params: UpdateInputNumberParams
) -> InputNumberRecord

update(
    helper_id: str, params: UpdateInputTextParams
) -> InputTextRecord

update(
    helper_id: str, params: UpdateInputSelectParams
) -> InputSelectRecord

update(
    helper_id: str, params: UpdateInputDatetimeParams
) -> InputDatetimeRecord

update(
    helper_id: str, params: UpdateInputButtonParams
) -> InputButtonRecord

update(
    helper_id: str, params: UpdateCounterParams
) -> CounterRecord

update(
    helper_id: str, params: UpdateTimerParams
) -> TimerRecord

Update an existing helper.

Parameters:

Name	Type	Description	Default
helper_id	str	The ID of the helper to update.	required
params	BaseModel	Fields to update (unset fields are left unchanged). The concrete type determines the domain and the return type via overload resolution.	required

Returns:

Type	Description
BaseModel	The updated stored record.

Source code in src/hassette/api/helpers.py

async def update(self, helper_id: str, params: BaseModel) -> BaseModel:
    """Update an existing helper.

    Args:
        helper_id: The ID of the helper to update.
        params: Fields to update (unset fields are left unchanged). The concrete type
            determines the domain and the return type via overload resolution.

    Returns:
        The updated stored record.
    """
    domain, record_type, id_key = UPDATE_DISPATCH[type(params)]
    val = await _ws_helper_call(
        self._api, domain, "update", **{id_key: helper_id}, **params.model_dump(exclude_unset=True)
    )
    record = record_type.model_validate(_expect_dict(val, f"{domain}/update"))
    self.logger.debug("Updated %s helper %r", domain, helper_id)
    return record

delete(domain: HelperDomain, helper_id: str) -> None async

delete(
    domain: Literal["input_boolean"], helper_id: str
) -> None

delete(
    domain: Literal["input_number"], helper_id: str
) -> None

delete(
    domain: Literal["input_text"], helper_id: str
) -> None

delete(
    domain: Literal["input_select"], helper_id: str
) -> None

delete(
    domain: Literal["input_datetime"], helper_id: str
) -> None

delete(
    domain: Literal["input_button"], helper_id: str
) -> None

delete(domain: Literal['counter'], helper_id: str) -> None

delete(domain: Literal['timer'], helper_id: str) -> None

Delete a helper.

Parameters:

Name	Type	Description	Default
domain	HelperDomain	The helper domain (e.g. "input_boolean", "counter").	required
helper_id	str	The ID of the helper to delete.	required

Source code in src/hassette/api/helpers.py

async def delete(self, domain: HelperDomain, helper_id: str) -> None:
    """Delete a helper.

    Args:
        domain: The helper domain (e.g. "input_boolean", "counter").
        helper_id: The ID of the helper to delete.
    """
    id_key = ID_KEYS[domain]
    await _ws_helper_call(self._api, domain, "delete", **{id_key: helper_id})
    self.logger.debug("Deleted %s helper %r", domain, helper_id)

increment(entity_id: str) -> None async

Increment a counter entity's current value (live state, not stored config).

Parameters:

Name	Type	Description	Default
entity_id	str	The entity ID of the counter (e.g. "counter.motion_count").	required

Source code in src/hassette/api/helpers.py

async def increment(self, entity_id: str) -> None:
    """Increment a counter entity's current value (live state, not stored config).

    Args:
        entity_id: The entity ID of the counter (e.g. ``"counter.motion_count"``).
    """
    await self._api.call_service(
        "counter",
        "increment",
        target={"entity_id": entity_id},
        return_response=True,  # surfaces HA errors instead of fire-and-forget
    )
    self.logger.debug("Incremented counter %r", entity_id)

decrement(entity_id: str) -> None async

Decrement a counter entity's current value (live state, not stored config).

Parameters:

Name	Type	Description	Default
entity_id	str	The entity ID of the counter (e.g. "counter.motion_count").	required

Source code in src/hassette/api/helpers.py

async def decrement(self, entity_id: str) -> None:
    """Decrement a counter entity's current value (live state, not stored config).

    Args:
        entity_id: The entity ID of the counter (e.g. ``"counter.motion_count"``).
    """
    await self._api.call_service(
        "counter",
        "decrement",
        target={"entity_id": entity_id},
        return_response=True,
    )
    self.logger.debug("Decremented counter %r", entity_id)

reset(entity_id: str) -> None async

Reset a counter entity's value to its configured initial (live state, not stored config).

Parameters:

Name	Type	Description	Default
entity_id	str	The entity ID of the counter (e.g. "counter.motion_count").	required

Source code in src/hassette/api/helpers.py

async def reset(self, entity_id: str) -> None:
    """Reset a counter entity's value to its configured initial (live state, not stored config).

    Args:
        entity_id: The entity ID of the counter (e.g. ``"counter.motion_count"``).
    """
    await self._api.call_service(
        "counter",
        "reset",
        target={"entity_id": entity_id},
        return_response=True,
    )
    self.logger.debug("Reset counter %r", entity_id)

ApiSyncFacade

Bases: Resource

Synchronous facade for the API service.

This class provides synchronous methods that wrap the asynchronous methods of the Api class, allowing for blocking calls in a synchronous context.

These methods must not be called from within the event loop; doing so raises a RuntimeError. Use the asynchronous methods on Api directly when operating within an event loop.

Source code in src/hassette/api/sync.py

class ApiSyncFacade(Resource):
    """Synchronous facade for the API service.

    This class provides synchronous methods that wrap the asynchronous methods of the Api class,
    allowing for blocking calls in a synchronous context.

    These methods must not be called from within the event loop; doing so raises a RuntimeError.
    Use the asynchronous methods on ``Api`` directly when operating within an event loop.
    """

    _api: "Api"

    def __init__(self, hassette: "Hassette", *, api: "Api", parent: Resource | None = None) -> None:
        super().__init__(hassette, parent=parent)
        self._api = api
        self.helpers = self.add_child(HelperClientSyncFacade, helpers=self._api.helpers)

    async def on_initialize(self) -> None:
        mark_ready(self, reason="Synchronous API facade initialized")

    @property
    def config_log_level(self) -> LOG_LEVEL_TYPE:
        return self.hassette.config.logging.api

    def ws_send_and_wait(self, **data: Any) -> Any:
        """Send a WebSocket message and wait for a response."""
        return self.task_bucket.run_sync(self._api.ws_send_and_wait(**data))

    def ws_send_json(self, **data: Any) -> None:
        """Send a WebSocket message without waiting for a response."""
        return self.task_bucket.run_sync(self._api.ws_send_json(**data))

    def rest_request(
        self,
        method: str,
        url: str,
        params: dict[str, Any] | None = None,
        data: dict[str, Any] | None = None,
        suppress_error_message: bool = False,
        **kwargs: Any,
    ) -> aiohttp.ClientResponse:
        """Make a REST request to the Home Assistant API.

        Args:
            method: The HTTP method to use (e.g., "GET", "POST").
            url: The URL endpoint for the request.
            params: Query parameters for the request.
            data: JSON payload for the request.
            suppress_error_message: Whether to suppress error messages.

        Returns:
            The response from the API.
        """
        return self.task_bucket.run_sync(
            self._api.rest_request(method, url, params, data, suppress_error_message, **kwargs)
        )

    def get_rest_request(self, url: str, params: dict[str, Any] | None = None, **kwargs: Any) -> aiohttp.ClientResponse:
        """Make a GET request to the Home Assistant API.

        Args:
            url: The URL endpoint for the request.
            params: Query parameters for the request.
            kwargs: Additional keyword arguments to pass to the request.

        Returns:
            The response from the API.
        """
        return self.task_bucket.run_sync(self._api.get_rest_request(url, params, **kwargs))

    def post_rest_request(self, url: str, data: dict[str, Any] | None = None, **kwargs: Any) -> aiohttp.ClientResponse:
        """Make a POST request to the Home Assistant API.

        Args:
            url: The URL endpoint for the request.
            data: JSON payload for the request.
            kwargs: Additional keyword arguments to pass to the request.

        Returns:
            The response from the API.
        """
        return self.task_bucket.run_sync(self._api.post_rest_request(url, data, **kwargs))

    def delete_rest_request(self, url: str, **kwargs: Any) -> aiohttp.ClientResponse:
        """Make a DELETE request to the Home Assistant API.

        Args:
            url: The URL endpoint for the request.
            kwargs: Additional keyword arguments to pass to the request.

        Returns:
            The response from the API.
        """
        return self.task_bucket.run_sync(self._api.delete_rest_request(url, **kwargs))

    def get_states_raw(self) -> list["HassStateDict"]:
        """Get all entities in Home Assistant as raw dictionaries.

        Returns:
            A list of states as dictionaries.
        """
        return self.task_bucket.run_sync(self._api.get_states_raw())

    def get_states(self) -> list["BaseState"]:
        """Get all entities in Home Assistant, converted to their appropriate state types.

        If a state fails to convert, it is skipped with an error logged. If there is no registered
        state class for a domain, the generic BaseState is used.

        Returns:
            A list of states, converted to their appropriate state types.
        """
        return self.task_bucket.run_sync(self._api.get_states())

    def get_config(self) -> dict[str, Any]:
        """Get the Home Assistant configuration.

        Returns:
            The configuration data.
        """
        return self.task_bucket.run_sync(self._api.get_config())

    def get_services(self) -> dict[str, Any]:
        """Get the available services in Home Assistant.

        Returns:
            The services data.
        """
        return self.task_bucket.run_sync(self._api.get_services())

    def get_panels(self) -> dict[str, Any]:
        """Get the available panels in Home Assistant.

        Returns:
            The panels data.
        """
        return self.task_bucket.run_sync(self._api.get_panels())

    def fire_event(self, event_type: str, event_data: dict[str, Any] | None = None) -> dict[str, Any]:
        """Fire a custom event in Home Assistant.

        Args:
            event_type: The type of the event to fire (e.g., "custom_event").
            event_data: Additional data to include with the event.

        Returns:
            The response from Home Assistant.
        """
        return self.task_bucket.run_sync(self._api.fire_event(event_type, event_data))

    def call_service(
        self,
        domain: str,
        service: str,
        target: dict[str, str] | dict[str, list[str]] | None = None,
        return_response: bool | None = False,
        **data: Any,
    ) -> ServiceResponse | None:
        """Call a Home Assistant service.

        Args:
            domain: The domain of the service (e.g., "light").
            service: The name of the service to call (e.g., "turn_on").
            target: Target entity IDs or areas.
            return_response: Whether to return the response from Home Assistant. Defaults to False.
            **data: Additional data to send with the service call.

        Returns:
            ServiceResponse | None: The response from Home Assistant if return_response is True. Otherwise None.
        """
        return self.task_bucket.run_sync(self._api.call_service(domain, service, target, return_response, **data))  # pyright: ignore[reportCallIssue, reportArgumentType]

    def turn_on(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None:
        """Turn on a specific entity in Home Assistant.

        Args:
            entity_id: The ID of the entity to turn on (e.g., "light.office").
            domain: The domain to use for the service call. Defaults to the entity's domain,
                derived from ``entity_id`` (e.g., ``"light.office"`` derives ``"light"``). Pass
                explicitly to override — for example, ``domain="homeassistant"`` to use the
                generic (deprecated) service.
            **data: Additional service data forwarded to the service call.
        """
        return self.task_bucket.run_sync(self._api.turn_on(entity_id, domain, **data))

    def turn_off(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None:
        """Turn off a specific entity in Home Assistant.

        Args:
            entity_id: The ID of the entity to turn off (e.g., "light.office").
            domain: The domain to use for the service call. Defaults to the entity's domain,
                derived from ``entity_id`` (e.g., ``"light.office"`` derives ``"light"``). Pass
                explicitly to override — for example, ``domain="homeassistant"`` to use the
                generic (deprecated) service.
            **data: Additional service data forwarded to the service call.
        """
        return self.task_bucket.run_sync(self._api.turn_off(entity_id, domain, **data))

    def toggle(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None:
        """Toggle a specific entity in Home Assistant.

        Args:
            entity_id: The ID of the entity to toggle (e.g., "light.office").
            domain: The domain to use for the service call. Defaults to the entity's domain,
                derived from ``entity_id`` (e.g., ``"light.office"`` derives ``"light"``). Pass
                explicitly to override — for example, ``domain="homeassistant"`` to use the
                generic (deprecated) service.
            **data: Additional service data forwarded to the service call.
        """
        return self.task_bucket.run_sync(self._api.toggle(entity_id, domain, **data))

    def get_state_raw(self, entity_id: str) -> "HassStateDict":
        """Get the state of a specific entity.

        Args:
            entity_id: The ID of the entity to get the state for.

        Returns:
            The state of the entity as raw data.
        """
        return self.task_bucket.run_sync(self._api.get_state_raw(entity_id))

    def entity_exists(self, entity_id: str) -> bool:
        """Check if a specific entity exists.

        Args:
            entity_id: The ID of the entity to check.

        Returns:
            True if the entity exists, False otherwise.
        """
        return self.task_bucket.run_sync(self._api.entity_exists(entity_id))

    def get_entity(self, entity_id: str, model: type["EntityT"]) -> "EntityT":
        """Get an entity object for a specific entity.

        Args:
            entity_id: The ID of the entity to get.
            model: The model class to use for the entity.

        Returns:
            The entity object.
        """
        return self.task_bucket.run_sync(self._api.get_entity(entity_id, model))

    def get_entity_or_none(self, entity_id: str, model: type["EntityT"]) -> "EntityT | None":
        """Get an entity object for a specific entity, or None if it does not exist.

        Args:
            entity_id: The ID of the entity to get.
            model: The model class to use for the entity.

        Returns:
            The entity object, or None if it does not exist.
        """
        return self.task_bucket.run_sync(self._api.get_entity_or_none(entity_id, model))

    def get_state(self, entity_id: str) -> "BaseState":
        """Get the state of a specific entity.

        Args:
            entity_id: The ID of the entity to get the state for.

        Returns:
            The state of the entity converted to the specified model type.
        """
        return self.task_bucket.run_sync(self._api.get_state(entity_id))

    def get_state_or_none(self, entity_id: str) -> "BaseState | None":
        """Get the state of a specific entity, or None if it does not exist.

        Args:
            entity_id: The ID of the entity to get the state for.

        Returns:
            The state of the entity converted to the specified model type, or None if it does not exist.
        """
        return self.task_bucket.run_sync(self._api.get_state_or_none(entity_id))

    def get_state_value(self, entity_id: str) -> Any:
        """Get the state of a specific entity without converting it to a state object.

        Args:
            entity_id: The ID of the entity to get the state for.

        Returns:
            The state of the entity as raw data.

        Note:
            While most default methods in this library work with state objects for
            strong typing, this method is designed to return the raw state value,
            as it is likely overkill to convert it to a state object for simple state value retrieval.
        """
        return self.task_bucket.run_sync(self._api.get_state_value(entity_id))

    def get_attribute(self, entity_id: str, attribute: str) -> Any | FalseySentinel:
        """Get a specific attribute of an entity.

        Args:
            entity_id: The ID of the entity to get the attribute for.
            attribute: The name of the attribute to retrieve. Can be a dot-separated path for nested attributes.

        Returns:
            The value of the specified attribute, or MISSING_VALUE sentinel if the attribute does not exist.
        """
        return self.task_bucket.run_sync(self._api.get_attribute(entity_id, attribute))

    def get_history(
        self,
        entity_id: str,
        start_time: PlainDateTime | ZonedDateTime | Date | str,
        end_time: PlainDateTime | ZonedDateTime | Date | str | None = None,
        significant_changes_only: bool = False,
        minimal_response: bool = False,
        no_attributes: bool = False,
    ) -> list[HistoryEntry]:
        """Get the history of a specific entity.

        Args:
            entity_id: The ID of the entity to get the history for.
            start_time: The start time for the history range.
            end_time: The end time for the history range.
            significant_changes_only: Whether to only include significant changes.
            minimal_response: Whether to request a minimal response.
            no_attributes: Whether to exclude attributes from the response.

        Returns:
            A list of history entries for the specified entity.
        """
        return self.task_bucket.run_sync(
            self._api.get_history(
                entity_id, start_time, end_time, significant_changes_only, minimal_response, no_attributes
            )
        )

    def get_histories(
        self,
        entity_ids: list[str],
        start_time: PlainDateTime | ZonedDateTime | Date | str,
        end_time: PlainDateTime | ZonedDateTime | Date | str | None = None,
        significant_changes_only: bool = False,
        minimal_response: bool = False,
        no_attributes: bool = False,
    ) -> dict[str, list[HistoryEntry]]:
        """Get the history for multiple entities.

        Args:
            entity_ids: The IDs of the entities to get the history for.
            start_time: The start time for the history range.
            end_time: The end time for the history range.
            significant_changes_only: Whether to only include significant changes.
            minimal_response: Whether to request a minimal response.
            no_attributes: Whether to exclude attributes from the response.

        Returns:
            A dictionary mapping entity IDs to their respective history entries.
        """
        return self.task_bucket.run_sync(
            self._api.get_histories(
                entity_ids, start_time, end_time, significant_changes_only, minimal_response, no_attributes
            )
        )

    def get_logbook(
        self,
        entity_id: str,
        start_time: PlainDateTime | ZonedDateTime | Date | str,
        end_time: PlainDateTime | ZonedDateTime | Date | str,
    ) -> list[dict[str, Any]]:
        """Get the logbook entries for a specific entity.

        Args:
            entity_id: The ID of the entity to get the logbook entries for.
            start_time: The start time for the logbook range.
            end_time: The end time for the logbook range.

        Returns:
            A list of logbook entries for the specified entity.
        """
        return self.task_bucket.run_sync(self._api.get_logbook(entity_id, start_time, end_time))

    def set_state(self, entity_id: str | StrEnum, state: Any, attributes: dict[str, Any] | None = None) -> dict:
        """Set the state of a specific entity.

        Args:
            entity_id: The ID of the entity to set the state for.
            state: The new state value to set.
            attributes: Additional attributes to set for the entity.

        Returns:
            The response from Home Assistant after setting the state.
        """
        return self.task_bucket.run_sync(self._api.set_state(entity_id, state, attributes))

    def get_camera_image(
        self, entity_id: str, timestamp: PlainDateTime | ZonedDateTime | Date | str | None = None
    ) -> bytes:
        """Get the latest camera image for a specific entity.

        Args:
            entity_id: The ID of the camera entity to get the image for.
            timestamp: The timestamp for the image. If None, the latest image is returned.

        Returns:
            The camera image data.
        """
        return self.task_bucket.run_sync(self._api.get_camera_image(entity_id, timestamp))

    def get_calendars(self) -> list[dict[str, Any]]:
        """Get the list of calendars."""
        return self.task_bucket.run_sync(self._api.get_calendars())

    def get_calendar_events(
        self,
        calendar_id: str,
        start_time: PlainDateTime | ZonedDateTime | Date | str,
        end_time: PlainDateTime | ZonedDateTime | Date | str,
    ) -> list[dict[str, Any]]:
        """Get events from a specific calendar.

        Args:
            calendar_id: The ID of the calendar to get events from.
            start_time: The start time for the event range.
            end_time: The end time for the event range.

        Returns:
            A list of calendar events.
        """
        return self.task_bucket.run_sync(self._api.get_calendar_events(calendar_id, start_time, end_time))

    def render_template(self, template: str, variables: dict[str, Any] | None = None) -> str:
        """Render a template with given variables.

        Args:
            template: The template string to render.
            variables: Variables to use in the template.

        Returns:
            The rendered template result.
        """
        return self.task_bucket.run_sync(self._api.render_template(template, variables))

    def delete_entity(self, entity_id: str) -> None:
        """Delete a specific entity.

        Args:
            entity_id: The ID of the entity to delete.

        Raises:
            RuntimeError: If the deletion fails.
        """
        return self.task_bucket.run_sync(self._api.delete_entity(entity_id))

ws_send_and_wait(**data: Any) -> Any

Send a WebSocket message and wait for a response.

Source code in src/hassette/api/sync.py

def ws_send_and_wait(self, **data: Any) -> Any:
    """Send a WebSocket message and wait for a response."""
    return self.task_bucket.run_sync(self._api.ws_send_and_wait(**data))

ws_send_json(**data: Any) -> None

Send a WebSocket message without waiting for a response.

Source code in src/hassette/api/sync.py

def ws_send_json(self, **data: Any) -> None:
    """Send a WebSocket message without waiting for a response."""
    return self.task_bucket.run_sync(self._api.ws_send_json(**data))

rest_request(method: str, url: str, params: dict[str, Any] | None = None, data: dict[str, Any] | None = None, suppress_error_message: bool = False, **kwargs: Any) -> aiohttp.ClientResponse

Make a REST request to the Home Assistant API.

Parameters:

Name	Type	Description	Default
method	str	The HTTP method to use (e.g., "GET", "POST").	required
url	str	The URL endpoint for the request.	required
params	dict[str, Any] | None	Query parameters for the request.	None
data	dict[str, Any] | None	JSON payload for the request.	None
suppress_error_message	bool	Whether to suppress error messages.	False

Returns:

Type	Description
ClientResponse	The response from the API.

Source code in src/hassette/api/sync.py

def rest_request(
    self,
    method: str,
    url: str,
    params: dict[str, Any] | None = None,
    data: dict[str, Any] | None = None,
    suppress_error_message: bool = False,
    **kwargs: Any,
) -> aiohttp.ClientResponse:
    """Make a REST request to the Home Assistant API.

    Args:
        method: The HTTP method to use (e.g., "GET", "POST").
        url: The URL endpoint for the request.
        params: Query parameters for the request.
        data: JSON payload for the request.
        suppress_error_message: Whether to suppress error messages.

    Returns:
        The response from the API.
    """
    return self.task_bucket.run_sync(
        self._api.rest_request(method, url, params, data, suppress_error_message, **kwargs)
    )

get_rest_request(url: str, params: dict[str, Any] | None = None, **kwargs: Any) -> aiohttp.ClientResponse

Make a GET request to the Home Assistant API.

Parameters:

Name	Type	Description	Default
url	str	The URL endpoint for the request.	required
params	dict[str, Any] | None	Query parameters for the request.	None
kwargs	Any	Additional keyword arguments to pass to the request.	{}

Returns:

Type	Description
ClientResponse	The response from the API.

Source code in src/hassette/api/sync.py

def get_rest_request(self, url: str, params: dict[str, Any] | None = None, **kwargs: Any) -> aiohttp.ClientResponse:
    """Make a GET request to the Home Assistant API.

    Args:
        url: The URL endpoint for the request.
        params: Query parameters for the request.
        kwargs: Additional keyword arguments to pass to the request.

    Returns:
        The response from the API.
    """
    return self.task_bucket.run_sync(self._api.get_rest_request(url, params, **kwargs))

post_rest_request(url: str, data: dict[str, Any] | None = None, **kwargs: Any) -> aiohttp.ClientResponse

Make a POST request to the Home Assistant API.

Parameters:

Name	Type	Description	Default
url	str	The URL endpoint for the request.	required
data	dict[str, Any] | None	JSON payload for the request.	None
kwargs	Any	Additional keyword arguments to pass to the request.	{}

Returns:

Type	Description
ClientResponse	The response from the API.

Source code in src/hassette/api/sync.py

def post_rest_request(self, url: str, data: dict[str, Any] | None = None, **kwargs: Any) -> aiohttp.ClientResponse:
    """Make a POST request to the Home Assistant API.

    Args:
        url: The URL endpoint for the request.
        data: JSON payload for the request.
        kwargs: Additional keyword arguments to pass to the request.

    Returns:
        The response from the API.
    """
    return self.task_bucket.run_sync(self._api.post_rest_request(url, data, **kwargs))

delete_rest_request(url: str, **kwargs: Any) -> aiohttp.ClientResponse

Make a DELETE request to the Home Assistant API.

Parameters:

Name	Type	Description	Default
url	str	The URL endpoint for the request.	required
kwargs	Any	Additional keyword arguments to pass to the request.	{}

Returns:

Type	Description
ClientResponse	The response from the API.

Source code in src/hassette/api/sync.py

def delete_rest_request(self, url: str, **kwargs: Any) -> aiohttp.ClientResponse:
    """Make a DELETE request to the Home Assistant API.

    Args:
        url: The URL endpoint for the request.
        kwargs: Additional keyword arguments to pass to the request.

    Returns:
        The response from the API.
    """
    return self.task_bucket.run_sync(self._api.delete_rest_request(url, **kwargs))

get_states_raw() -> list[HassStateDict]

Get all entities in Home Assistant as raw dictionaries.

Returns:

Type	Description
list[HassStateDict]	A list of states as dictionaries.

Source code in src/hassette/api/sync.py

def get_states_raw(self) -> list["HassStateDict"]:
    """Get all entities in Home Assistant as raw dictionaries.

    Returns:
        A list of states as dictionaries.
    """
    return self.task_bucket.run_sync(self._api.get_states_raw())

get_states() -> list[BaseState]

Get all entities in Home Assistant, converted to their appropriate state types.

If a state fails to convert, it is skipped with an error logged. If there is no registered state class for a domain, the generic BaseState is used.

Returns:

Type	Description
list[BaseState]	A list of states, converted to their appropriate state types.

Source code in src/hassette/api/sync.py

def get_states(self) -> list["BaseState"]:
    """Get all entities in Home Assistant, converted to their appropriate state types.

    If a state fails to convert, it is skipped with an error logged. If there is no registered
    state class for a domain, the generic BaseState is used.

    Returns:
        A list of states, converted to their appropriate state types.
    """
    return self.task_bucket.run_sync(self._api.get_states())

get_config() -> dict[str, Any]

Get the Home Assistant configuration.

Returns:

Type	Description
dict[str, Any]	The configuration data.

Source code in src/hassette/api/sync.py

def get_config(self) -> dict[str, Any]:
    """Get the Home Assistant configuration.

    Returns:
        The configuration data.
    """
    return self.task_bucket.run_sync(self._api.get_config())

get_services() -> dict[str, Any]

Get the available services in Home Assistant.

Returns:

Type	Description
dict[str, Any]	The services data.

Source code in src/hassette/api/sync.py

def get_services(self) -> dict[str, Any]:
    """Get the available services in Home Assistant.

    Returns:
        The services data.
    """
    return self.task_bucket.run_sync(self._api.get_services())

get_panels() -> dict[str, Any]

Get the available panels in Home Assistant.

Returns:

Type	Description
dict[str, Any]	The panels data.

Source code in src/hassette/api/sync.py

def get_panels(self) -> dict[str, Any]:
    """Get the available panels in Home Assistant.

    Returns:
        The panels data.
    """
    return self.task_bucket.run_sync(self._api.get_panels())

fire_event(event_type: str, event_data: dict[str, Any] | None = None) -> dict[str, Any]

Fire a custom event in Home Assistant.

Parameters:

Name	Type	Description	Default
event_type	str	The type of the event to fire (e.g., "custom_event").	required
event_data	dict[str, Any] | None	Additional data to include with the event.	None

Returns:

Type	Description
dict[str, Any]	The response from Home Assistant.

Source code in src/hassette/api/sync.py

def fire_event(self, event_type: str, event_data: dict[str, Any] | None = None) -> dict[str, Any]:
    """Fire a custom event in Home Assistant.

    Args:
        event_type: The type of the event to fire (e.g., "custom_event").
        event_data: Additional data to include with the event.

    Returns:
        The response from Home Assistant.
    """
    return self.task_bucket.run_sync(self._api.fire_event(event_type, event_data))

call_service(domain: str, service: str, target: dict[str, str] | dict[str, list[str]] | None = None, return_response: bool | None = False, **data: Any) -> ServiceResponse | None

Call a Home Assistant service.

Parameters:

Name	Type	Description	Default
domain	str	The domain of the service (e.g., "light").	required
service	str	The name of the service to call (e.g., "turn_on").	required
target	dict[str, str] | dict[str, list[str]] | None	Target entity IDs or areas.	None
return_response	bool | None	Whether to return the response from Home Assistant. Defaults to False.	False
**data	Any	Additional data to send with the service call.	{}

Returns:

Type	Description
ServiceResponse | None	ServiceResponse | None: The response from Home Assistant if return_response is True. Otherwise None.

Source code in src/hassette/api/sync.py

def call_service(
    self,
    domain: str,
    service: str,
    target: dict[str, str] | dict[str, list[str]] | None = None,
    return_response: bool | None = False,
    **data: Any,
) -> ServiceResponse | None:
    """Call a Home Assistant service.

    Args:
        domain: The domain of the service (e.g., "light").
        service: The name of the service to call (e.g., "turn_on").
        target: Target entity IDs or areas.
        return_response: Whether to return the response from Home Assistant. Defaults to False.
        **data: Additional data to send with the service call.

    Returns:
        ServiceResponse | None: The response from Home Assistant if return_response is True. Otherwise None.
    """
    return self.task_bucket.run_sync(self._api.call_service(domain, service, target, return_response, **data))  # pyright: ignore[reportCallIssue, reportArgumentType]

turn_on(entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None

Turn on a specific entity in Home Assistant.

Parameters:

Name	Type	Description	Default
entity_id	str | StrEnum	The ID of the entity to turn on (e.g., "light.office").	required
domain	str | None	The domain to use for the service call. Defaults to the entity's domain, derived from entity_id (e.g., "light.office" derives "light"). Pass explicitly to override — for example, domain="homeassistant" to use the generic (deprecated) service.	None
**data	Any	Additional service data forwarded to the service call.	{}

Source code in src/hassette/api/sync.py

def turn_on(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None:
    """Turn on a specific entity in Home Assistant.

    Args:
        entity_id: The ID of the entity to turn on (e.g., "light.office").
        domain: The domain to use for the service call. Defaults to the entity's domain,
            derived from ``entity_id`` (e.g., ``"light.office"`` derives ``"light"``). Pass
            explicitly to override — for example, ``domain="homeassistant"`` to use the
            generic (deprecated) service.
        **data: Additional service data forwarded to the service call.
    """
    return self.task_bucket.run_sync(self._api.turn_on(entity_id, domain, **data))

turn_off(entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None

Turn off a specific entity in Home Assistant.

Parameters:

Name	Type	Description	Default
entity_id	str | StrEnum	The ID of the entity to turn off (e.g., "light.office").	required
domain	str | None	The domain to use for the service call. Defaults to the entity's domain, derived from entity_id (e.g., "light.office" derives "light"). Pass explicitly to override — for example, domain="homeassistant" to use the generic (deprecated) service.	None
**data	Any	Additional service data forwarded to the service call.	{}

Source code in src/hassette/api/sync.py

def turn_off(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None:
    """Turn off a specific entity in Home Assistant.

    Args:
        entity_id: The ID of the entity to turn off (e.g., "light.office").
        domain: The domain to use for the service call. Defaults to the entity's domain,
            derived from ``entity_id`` (e.g., ``"light.office"`` derives ``"light"``). Pass
            explicitly to override — for example, ``domain="homeassistant"`` to use the
            generic (deprecated) service.
        **data: Additional service data forwarded to the service call.
    """
    return self.task_bucket.run_sync(self._api.turn_off(entity_id, domain, **data))

toggle(entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None

Toggle a specific entity in Home Assistant.

Parameters:

Name	Type	Description	Default
entity_id	str | StrEnum	The ID of the entity to toggle (e.g., "light.office").	required
domain	str | None	The domain to use for the service call. Defaults to the entity's domain, derived from entity_id (e.g., "light.office" derives "light"). Pass explicitly to override — for example, domain="homeassistant" to use the generic (deprecated) service.	None
**data	Any	Additional service data forwarded to the service call.	{}

Source code in src/hassette/api/sync.py

def toggle(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None:
    """Toggle a specific entity in Home Assistant.

    Args:
        entity_id: The ID of the entity to toggle (e.g., "light.office").
        domain: The domain to use for the service call. Defaults to the entity's domain,
            derived from ``entity_id`` (e.g., ``"light.office"`` derives ``"light"``). Pass
            explicitly to override — for example, ``domain="homeassistant"`` to use the
            generic (deprecated) service.
        **data: Additional service data forwarded to the service call.
    """
    return self.task_bucket.run_sync(self._api.toggle(entity_id, domain, **data))

get_state_raw(entity_id: str) -> HassStateDict

Get the state of a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get the state for.	required

Returns:

Type	Description
HassStateDict	The state of the entity as raw data.

Source code in src/hassette/api/sync.py

def get_state_raw(self, entity_id: str) -> "HassStateDict":
    """Get the state of a specific entity.

    Args:
        entity_id: The ID of the entity to get the state for.

    Returns:
        The state of the entity as raw data.
    """
    return self.task_bucket.run_sync(self._api.get_state_raw(entity_id))

entity_exists(entity_id: str) -> bool

Check if a specific entity exists.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to check.	required

Returns:

Type	Description
bool	True if the entity exists, False otherwise.

Source code in src/hassette/api/sync.py

def entity_exists(self, entity_id: str) -> bool:
    """Check if a specific entity exists.

    Args:
        entity_id: The ID of the entity to check.

    Returns:
        True if the entity exists, False otherwise.
    """
    return self.task_bucket.run_sync(self._api.entity_exists(entity_id))

get_entity(entity_id: str, model: type[EntityT]) -> EntityT

Get an entity object for a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get.	required
model	type[EntityT]	The model class to use for the entity.	required

Returns:

Type	Description
EntityT	The entity object.

Source code in src/hassette/api/sync.py

def get_entity(self, entity_id: str, model: type["EntityT"]) -> "EntityT":
    """Get an entity object for a specific entity.

    Args:
        entity_id: The ID of the entity to get.
        model: The model class to use for the entity.

    Returns:
        The entity object.
    """
    return self.task_bucket.run_sync(self._api.get_entity(entity_id, model))

get_entity_or_none(entity_id: str, model: type[EntityT]) -> EntityT | None

Get an entity object for a specific entity, or None if it does not exist.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get.	required
model	type[EntityT]	The model class to use for the entity.	required

Returns:

Type	Description
EntityT | None	The entity object, or None if it does not exist.

Source code in src/hassette/api/sync.py

def get_entity_or_none(self, entity_id: str, model: type["EntityT"]) -> "EntityT | None":
    """Get an entity object for a specific entity, or None if it does not exist.

    Args:
        entity_id: The ID of the entity to get.
        model: The model class to use for the entity.

    Returns:
        The entity object, or None if it does not exist.
    """
    return self.task_bucket.run_sync(self._api.get_entity_or_none(entity_id, model))

get_state(entity_id: str) -> BaseState

Get the state of a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get the state for.	required

Returns:

Type	Description
BaseState	The state of the entity converted to the specified model type.

Source code in src/hassette/api/sync.py

def get_state(self, entity_id: str) -> "BaseState":
    """Get the state of a specific entity.

    Args:
        entity_id: The ID of the entity to get the state for.

    Returns:
        The state of the entity converted to the specified model type.
    """
    return self.task_bucket.run_sync(self._api.get_state(entity_id))

get_state_or_none(entity_id: str) -> BaseState | None

Get the state of a specific entity, or None if it does not exist.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get the state for.	required

Returns:

Type	Description
BaseState | None	The state of the entity converted to the specified model type, or None if it does not exist.

Source code in src/hassette/api/sync.py

def get_state_or_none(self, entity_id: str) -> "BaseState | None":
    """Get the state of a specific entity, or None if it does not exist.

    Args:
        entity_id: The ID of the entity to get the state for.

    Returns:
        The state of the entity converted to the specified model type, or None if it does not exist.
    """
    return self.task_bucket.run_sync(self._api.get_state_or_none(entity_id))

get_state_value(entity_id: str) -> Any

Get the state of a specific entity without converting it to a state object.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get the state for.	required

Returns:

Type	Description
Any	The state of the entity as raw data.

Note

While most default methods in this library work with state objects for strong typing, this method is designed to return the raw state value, as it is likely overkill to convert it to a state object for simple state value retrieval.

Source code in src/hassette/api/sync.py

def get_state_value(self, entity_id: str) -> Any:
    """Get the state of a specific entity without converting it to a state object.

    Args:
        entity_id: The ID of the entity to get the state for.

    Returns:
        The state of the entity as raw data.

    Note:
        While most default methods in this library work with state objects for
        strong typing, this method is designed to return the raw state value,
        as it is likely overkill to convert it to a state object for simple state value retrieval.
    """
    return self.task_bucket.run_sync(self._api.get_state_value(entity_id))

get_attribute(entity_id: str, attribute: str) -> Any | FalseySentinel

Get a specific attribute of an entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get the attribute for.	required
attribute	str	The name of the attribute to retrieve. Can be a dot-separated path for nested attributes.	required

Returns:

Type	Description
Any | FalseySentinel	The value of the specified attribute, or MISSING_VALUE sentinel if the attribute does not exist.

Source code in src/hassette/api/sync.py

def get_attribute(self, entity_id: str, attribute: str) -> Any | FalseySentinel:
    """Get a specific attribute of an entity.

    Args:
        entity_id: The ID of the entity to get the attribute for.
        attribute: The name of the attribute to retrieve. Can be a dot-separated path for nested attributes.

    Returns:
        The value of the specified attribute, or MISSING_VALUE sentinel if the attribute does not exist.
    """
    return self.task_bucket.run_sync(self._api.get_attribute(entity_id, attribute))

get_history(entity_id: str, start_time: PlainDateTime | ZonedDateTime | Date | str, end_time: PlainDateTime | ZonedDateTime | Date | str | None = None, significant_changes_only: bool = False, minimal_response: bool = False, no_attributes: bool = False) -> list[HistoryEntry]

Get the history of a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get the history for.	required
start_time	PlainDateTime | ZonedDateTime | Date | str	The start time for the history range.	required
end_time	PlainDateTime | ZonedDateTime | Date | str | None	The end time for the history range.	None
significant_changes_only	bool	Whether to only include significant changes.	False
minimal_response	bool	Whether to request a minimal response.	False
no_attributes	bool	Whether to exclude attributes from the response.	False

Returns:

Type	Description
list[HistoryEntry]	A list of history entries for the specified entity.

Source code in src/hassette/api/sync.py

def get_history(
    self,
    entity_id: str,
    start_time: PlainDateTime | ZonedDateTime | Date | str,
    end_time: PlainDateTime | ZonedDateTime | Date | str | None = None,
    significant_changes_only: bool = False,
    minimal_response: bool = False,
    no_attributes: bool = False,
) -> list[HistoryEntry]:
    """Get the history of a specific entity.

    Args:
        entity_id: The ID of the entity to get the history for.
        start_time: The start time for the history range.
        end_time: The end time for the history range.
        significant_changes_only: Whether to only include significant changes.
        minimal_response: Whether to request a minimal response.
        no_attributes: Whether to exclude attributes from the response.

    Returns:
        A list of history entries for the specified entity.
    """
    return self.task_bucket.run_sync(
        self._api.get_history(
            entity_id, start_time, end_time, significant_changes_only, minimal_response, no_attributes
        )
    )

get_histories(entity_ids: list[str], start_time: PlainDateTime | ZonedDateTime | Date | str, end_time: PlainDateTime | ZonedDateTime | Date | str | None = None, significant_changes_only: bool = False, minimal_response: bool = False, no_attributes: bool = False) -> dict[str, list[HistoryEntry]]

Get the history for multiple entities.

Parameters:

Name	Type	Description	Default
entity_ids	list[str]	The IDs of the entities to get the history for.	required
start_time	PlainDateTime | ZonedDateTime | Date | str	The start time for the history range.	required
end_time	PlainDateTime | ZonedDateTime | Date | str | None	The end time for the history range.	None
significant_changes_only	bool	Whether to only include significant changes.	False
minimal_response	bool	Whether to request a minimal response.	False
no_attributes	bool	Whether to exclude attributes from the response.	False

Returns:

Type	Description
dict[str, list[HistoryEntry]]	A dictionary mapping entity IDs to their respective history entries.

Source code in src/hassette/api/sync.py

def get_histories(
    self,
    entity_ids: list[str],
    start_time: PlainDateTime | ZonedDateTime | Date | str,
    end_time: PlainDateTime | ZonedDateTime | Date | str | None = None,
    significant_changes_only: bool = False,
    minimal_response: bool = False,
    no_attributes: bool = False,
) -> dict[str, list[HistoryEntry]]:
    """Get the history for multiple entities.

    Args:
        entity_ids: The IDs of the entities to get the history for.
        start_time: The start time for the history range.
        end_time: The end time for the history range.
        significant_changes_only: Whether to only include significant changes.
        minimal_response: Whether to request a minimal response.
        no_attributes: Whether to exclude attributes from the response.

    Returns:
        A dictionary mapping entity IDs to their respective history entries.
    """
    return self.task_bucket.run_sync(
        self._api.get_histories(
            entity_ids, start_time, end_time, significant_changes_only, minimal_response, no_attributes
        )
    )

get_logbook(entity_id: str, start_time: PlainDateTime | ZonedDateTime | Date | str, end_time: PlainDateTime | ZonedDateTime | Date | str) -> list[dict[str, Any]]

Get the logbook entries for a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to get the logbook entries for.	required
start_time	PlainDateTime | ZonedDateTime | Date | str	The start time for the logbook range.	required
end_time	PlainDateTime | ZonedDateTime | Date | str	The end time for the logbook range.	required

Returns:

Type	Description
list[dict[str, Any]]	A list of logbook entries for the specified entity.

Source code in src/hassette/api/sync.py

def get_logbook(
    self,
    entity_id: str,
    start_time: PlainDateTime | ZonedDateTime | Date | str,
    end_time: PlainDateTime | ZonedDateTime | Date | str,
) -> list[dict[str, Any]]:
    """Get the logbook entries for a specific entity.

    Args:
        entity_id: The ID of the entity to get the logbook entries for.
        start_time: The start time for the logbook range.
        end_time: The end time for the logbook range.

    Returns:
        A list of logbook entries for the specified entity.
    """
    return self.task_bucket.run_sync(self._api.get_logbook(entity_id, start_time, end_time))

set_state(entity_id: str | StrEnum, state: Any, attributes: dict[str, Any] | None = None) -> dict

Set the state of a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str | StrEnum	The ID of the entity to set the state for.	required
state	Any	The new state value to set.	required
attributes	dict[str, Any] | None	Additional attributes to set for the entity.	None

Returns:

Type	Description
dict	The response from Home Assistant after setting the state.

Source code in src/hassette/api/sync.py

def set_state(self, entity_id: str | StrEnum, state: Any, attributes: dict[str, Any] | None = None) -> dict:
    """Set the state of a specific entity.

    Args:
        entity_id: The ID of the entity to set the state for.
        state: The new state value to set.
        attributes: Additional attributes to set for the entity.

    Returns:
        The response from Home Assistant after setting the state.
    """
    return self.task_bucket.run_sync(self._api.set_state(entity_id, state, attributes))

get_camera_image(entity_id: str, timestamp: PlainDateTime | ZonedDateTime | Date | str | None = None) -> bytes

Get the latest camera image for a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the camera entity to get the image for.	required
timestamp	PlainDateTime | ZonedDateTime | Date | str | None	The timestamp for the image. If None, the latest image is returned.	None

Returns:

Type	Description
bytes	The camera image data.

Source code in src/hassette/api/sync.py

def get_camera_image(
    self, entity_id: str, timestamp: PlainDateTime | ZonedDateTime | Date | str | None = None
) -> bytes:
    """Get the latest camera image for a specific entity.

    Args:
        entity_id: The ID of the camera entity to get the image for.
        timestamp: The timestamp for the image. If None, the latest image is returned.

    Returns:
        The camera image data.
    """
    return self.task_bucket.run_sync(self._api.get_camera_image(entity_id, timestamp))

get_calendars() -> list[dict[str, Any]]

Get the list of calendars.

Source code in src/hassette/api/sync.py

def get_calendars(self) -> list[dict[str, Any]]:
    """Get the list of calendars."""
    return self.task_bucket.run_sync(self._api.get_calendars())

get_calendar_events(calendar_id: str, start_time: PlainDateTime | ZonedDateTime | Date | str, end_time: PlainDateTime | ZonedDateTime | Date | str) -> list[dict[str, Any]]

Get events from a specific calendar.

Parameters:

Name	Type	Description	Default
calendar_id	str	The ID of the calendar to get events from.	required
start_time	PlainDateTime | ZonedDateTime | Date | str	The start time for the event range.	required
end_time	PlainDateTime | ZonedDateTime | Date | str	The end time for the event range.	required

Returns:

Type	Description
list[dict[str, Any]]	A list of calendar events.

Source code in src/hassette/api/sync.py

def get_calendar_events(
    self,
    calendar_id: str,
    start_time: PlainDateTime | ZonedDateTime | Date | str,
    end_time: PlainDateTime | ZonedDateTime | Date | str,
) -> list[dict[str, Any]]:
    """Get events from a specific calendar.

    Args:
        calendar_id: The ID of the calendar to get events from.
        start_time: The start time for the event range.
        end_time: The end time for the event range.

    Returns:
        A list of calendar events.
    """
    return self.task_bucket.run_sync(self._api.get_calendar_events(calendar_id, start_time, end_time))

render_template(template: str, variables: dict[str, Any] | None = None) -> str

Render a template with given variables.

Parameters:

Name	Type	Description	Default
template	str	The template string to render.	required
variables	dict[str, Any] | None	Variables to use in the template.	None

Returns:

Type	Description
str	The rendered template result.

Source code in src/hassette/api/sync.py

def render_template(self, template: str, variables: dict[str, Any] | None = None) -> str:
    """Render a template with given variables.

    Args:
        template: The template string to render.
        variables: Variables to use in the template.

    Returns:
        The rendered template result.
    """
    return self.task_bucket.run_sync(self._api.render_template(template, variables))

delete_entity(entity_id: str) -> None

Delete a specific entity.

Parameters:

Name	Type	Description	Default
entity_id	str	The ID of the entity to delete.	required

Raises:

Type	Description
RuntimeError	If the deletion fails.

Source code in src/hassette/api/sync.py

def delete_entity(self, entity_id: str) -> None:
    """Delete a specific entity.

    Args:
        entity_id: The ID of the entity to delete.

    Raises:
        RuntimeError: If the deletion fails.
    """
    return self.task_bucket.run_sync(self._api.delete_entity(entity_id))