Accessors

Accessors are combined with predicates and annotations to easily and cleanly extract values from events. Instead of writing a lambda like lambda e: e.payload.data.old_state_value, you can use the accessor get_state_value_old or use get_path("payload.data.old_state_value") for a more generic solution.

You generally will not need to use these directly - the main bus helpers use them under the hood to provide the relevant data to predicates. For example, on_state_change uses get_state_value_old and get_state_value_new to provide the old and new state values to predicates like StateFrom and StateTo.

Examples:

Extracting a specific key from service_data

from hassette import A
from hassette import P

value_is = P.ValueIs(source=A.get_service_data_key("entity_id"), condition="light.living_room")

await self.bus.on_call_service(
    "light.turn_on",
    handler=handler,
    where=value_is,
    name="living_room_turn_on",
)

Extracting a nested value using a glom path

from hassette import A
from hassette import P

value_is = P.ValueIs(
    source=A.get_path("payload.data.new_state.attributes.geolocation.locality"),
    condition="San Francisco",
)

await self.bus.on_state_change(
    "sensor.my_device_location",
    handler=handler,
    changed_to=value_is,
    name="device_location",
)

get_path(path: str) -> Callable[..., Any | FalseySentinel]

Return a callable that extracts a nested value, returning MISSING_VALUE on failure.

Source code in src/hassette/event_handling/accessors.py

def get_path(path: str) -> Callable[..., Any | FalseySentinel]:
    """Return a callable that extracts a nested value, returning MISSING_VALUE on failure."""

    def extract_path(obj):
        try:
            return glom(obj, path)
        except PathAccessError:
            # no logging for regular PathAccessError; just return MISSING_VALUE
            return MISSING_VALUE
        except Exception as exc:
            LOGGER.error("Error accessing path %r: %s - %s", path, type(exc).__name__, exc)
            return MISSING_VALUE

    return extract_path

get_state_value_old(event: RawStateChangeEvent) -> Any | FalseySentinel

Get the old state value from a RawStateChangeEvent, or MISSING_VALUE if old_state is None.

Source code in src/hassette/event_handling/accessors.py

def get_state_value_old(event: "RawStateChangeEvent") -> Any | FalseySentinel:
    """Get the old state value from a RawStateChangeEvent, or MISSING_VALUE if `old_state` is `None`."""
    return event.payload.data.old_state_value

get_state_value_new(event: RawStateChangeEvent) -> Any | FalseySentinel

Get the new state value from a RawStateChangeEvent, or MISSING_VALUE if new_state is None.

Source code in src/hassette/event_handling/accessors.py

def get_state_value_new(event: "RawStateChangeEvent") -> Any | FalseySentinel:
    """Get the new state value from a RawStateChangeEvent, or MISSING_VALUE if `new_state` is `None`."""
    return event.payload.data.new_state_value

get_state_value_old_new(event: RawStateChangeEvent) -> tuple[Any, Any]

Get a tuple of (old_state_value, new_state_value) from a RawStateChangeEvent.

Source code in src/hassette/event_handling/accessors.py

def get_state_value_old_new(event: "RawStateChangeEvent") -> tuple[Any, Any]:
    """Get a tuple of (old_state_value, new_state_value) from a RawStateChangeEvent."""
    return get_state_value_old(event), get_state_value_new(event)

get_state_object_old(event: RawStateChangeEvent) -> HassStateDict | None

Get the old state object from a RawStateChangeEvent, or None if old_state does not exist.

Source code in src/hassette/event_handling/accessors.py

def get_state_object_old(event: "RawStateChangeEvent") -> "HassStateDict | None":
    """Get the old state object from a RawStateChangeEvent, or None if `old_state` does not exist."""
    return event.payload.data.old_state

get_state_object_new(event: RawStateChangeEvent) -> HassStateDict | None

Get the new state object from a RawStateChangeEvent, or None if new_state does not exist.

Source code in src/hassette/event_handling/accessors.py

def get_state_object_new(event: "RawStateChangeEvent") -> "HassStateDict | None":
    """Get the new state object from a RawStateChangeEvent, or None if `new_state` does not exist."""
    return event.payload.data.new_state

get_state_object_old_new(event: RawStateChangeEvent) -> tuple[HassStateDict | None, HassStateDict | None]

Get a tuple of (old_state_object, new_state_object) from a RawStateChangeEvent.

Source code in src/hassette/event_handling/accessors.py

def get_state_object_old_new(event: "RawStateChangeEvent") -> tuple["HassStateDict | None", "HassStateDict | None"]:
    """Get a tuple of (old_state_object, new_state_object) from a RawStateChangeEvent."""
    return get_state_object_old(event), get_state_object_new(event)

get_attr_old(name: str) -> Callable[[RawStateChangeEvent], Any]

Get a specific attribute from the old state in a RawStateChangeEvent.

Source code in src/hassette/event_handling/accessors.py

def get_attr_old(name: str) -> Callable[["RawStateChangeEvent"], Any]:
    """Get a specific attribute from the old state in a RawStateChangeEvent."""

    def _inner(event: "RawStateChangeEvent") -> Any:
        data = event.payload.data
        old_attrs: dict[str, Any] = data.old_state.get("attributes", {}) if data.old_state else {}
        return old_attrs.get(name, MISSING_VALUE)

    return _inner

get_attr_new(name: str) -> Callable[[RawStateChangeEvent], Any]

Get a specific attribute from the new state in a RawStateChangeEvent.

Source code in src/hassette/event_handling/accessors.py

def get_attr_new(name: str) -> Callable[["RawStateChangeEvent"], Any]:
    """Get a specific attribute from the new state in a RawStateChangeEvent."""

    def _inner(event: "RawStateChangeEvent") -> Any:
        data = event.payload.data
        new_attrs: dict[str, Any] = data.new_state.get("attributes", {}) if data.new_state else {}
        return new_attrs.get(name, MISSING_VALUE)

    return _inner

get_attr_old_new(name: str) -> Callable[[RawStateChangeEvent], tuple[Any, Any]]

Get a specific attribute from the old and new state in a RawStateChangeEvent.

Source code in src/hassette/event_handling/accessors.py

def get_attr_old_new(name: str) -> Callable[["RawStateChangeEvent"], tuple[Any, Any]]:
    """Get a specific attribute from the old and new state in a RawStateChangeEvent."""
    old_getter = get_attr_old(name)
    new_getter = get_attr_new(name)

    def _inner(event: "RawStateChangeEvent") -> tuple[Any, Any]:
        old = old_getter(event)
        new = new_getter(event)
        return (old, new)

    return _inner

get_attrs_new(names: list[str]) -> Callable[[RawStateChangeEvent], dict[str, Any]]

Get specific attributes from the new state in a RawStateChangeEvent.

Source code in src/hassette/event_handling/accessors.py

def get_attrs_new(names: list[str]) -> Callable[["RawStateChangeEvent"], dict[str, Any]]:
    """Get specific attributes from the new state in a RawStateChangeEvent."""

    def _inner(event: "RawStateChangeEvent") -> dict[str, Any]:
        data = event.payload.data
        new_attrs: dict[str, Any] = data.new_state.get("attributes", {}) if data.new_state else {}
        return {name: new_attrs.get(name, MISSING_VALUE) for name in names}

    return _inner

get_attrs_old(names: list[str]) -> Callable[[RawStateChangeEvent], dict[str, Any]]

Get specific attributes from the old state in a RawStateChangeEvent.

Source code in src/hassette/event_handling/accessors.py

def get_attrs_old(names: list[str]) -> Callable[["RawStateChangeEvent"], dict[str, Any]]:
    """Get specific attributes from the old state in a RawStateChangeEvent."""

    def _inner(event: "RawStateChangeEvent") -> dict[str, Any]:
        data = event.payload.data
        old_attrs: dict[str, Any] = data.old_state.get("attributes", {}) if data.old_state else {}
        return {name: old_attrs.get(name, MISSING_VALUE) for name in names}

    return _inner

get_attrs_old_new(names: list[str]) -> Callable[[RawStateChangeEvent], tuple[dict[str, Any], dict[str, Any]]]

Get specific attributes from the old and new state in a RawStateChangeEvent.

Source code in src/hassette/event_handling/accessors.py

def get_attrs_old_new(
    names: list[str],
) -> Callable[["RawStateChangeEvent"], tuple[dict[str, Any], dict[str, Any]]]:
    """Get specific attributes from the old and new state in a RawStateChangeEvent."""
    old_getter = get_attrs_old(names)
    new_getter = get_attrs_new(names)

    def _inner(event: "RawStateChangeEvent") -> tuple[dict[str, Any], dict[str, Any]]:
        old = old_getter(event)
        new = new_getter(event)
        return (old, new)

    return _inner

get_all_attrs_old(event: RawStateChangeEvent) -> dict[str, Any] | FalseySentinel

Get all attributes from the old state in a RawStateChangeEvent.

Source code in src/hassette/event_handling/accessors.py

def get_all_attrs_old(event: "RawStateChangeEvent") -> dict[str, Any] | FalseySentinel:
    """Get all attributes from the old state in a RawStateChangeEvent."""
    data = event.payload.data
    if data.old_state is None:
        return MISSING_VALUE

    return data.old_state.get("attributes", {})

get_all_attrs_new(event: RawStateChangeEvent) -> dict[str, Any] | FalseySentinel

Get all attributes from the new state in a RawStateChangeEvent.

Source code in src/hassette/event_handling/accessors.py

def get_all_attrs_new(event: "RawStateChangeEvent") -> dict[str, Any] | FalseySentinel:
    """Get all attributes from the new state in a RawStateChangeEvent."""
    data = event.payload.data
    if data.new_state is None:
        return MISSING_VALUE

    return data.new_state.get("attributes", {})

get_all_attrs_old_new(event: RawStateChangeEvent) -> tuple[dict[str, Any] | FalseySentinel, dict[str, Any] | FalseySentinel]

Get all attributes from the old and new state in a RawStateChangeEvent.

Source code in src/hassette/event_handling/accessors.py

def get_all_attrs_old_new(
    event: "RawStateChangeEvent",
) -> tuple[dict[str, Any] | FalseySentinel, dict[str, Any] | FalseySentinel]:
    """Get all attributes from the old and new state in a RawStateChangeEvent."""
    old = get_all_attrs_old(event)
    new = get_all_attrs_new(event)
    return (old, new)

get_domain(event: HassEvent) -> str | FalseySentinel

Get the domain from the event payload.

Source code in src/hassette/event_handling/accessors.py

def get_domain(event: "HassEvent") -> str | FalseySentinel:
    """Get the domain from the event payload."""
    domain = cast("str", get_path("payload.data.domain")(event))
    if domain is not MISSING_VALUE:
        return domain

    if isinstance(event, RawStateChangeEvent):
        return event.payload.domain or MISSING_VALUE

    entity_id = get_entity_id(event)
    if isinstance(entity_id, str):
        return entity_id.split(".")[0]

    return MISSING_VALUE

get_entity_id(event: HassEvent) -> str | FalseySentinel

Get the entity_id from the event payload.

Source code in src/hassette/event_handling/accessors.py

def get_entity_id(event: "HassEvent") -> str | FalseySentinel:
    """Get the entity_id from the event payload."""
    entity_id = cast("str", get_path("payload.data.entity_id")(event))
    if entity_id is not MISSING_VALUE:
        return entity_id

    if isinstance(event, CallServiceEvent):
        return event.payload.data.service_data.get("entity_id", MISSING_VALUE)

    return MISSING_VALUE

get_context(event: HassEvent) -> HassContext

Get the context dict from the event payload.

Source code in src/hassette/event_handling/accessors.py

def get_context(event: "HassEvent") -> "HassContext":
    """Get the context dict from the event payload."""
    return cast("HassContext", get_path("payload.context")(event))

get_all_changes(exclude: Sequence[str] = DEFAULT_EXCLUDE) -> Callable[[RawStateChangeEvent], dict[str, dict[str, Any]]]

Get a dict of changed state and attribute values between old and new states.

Parameters:

Name	Type	Description	Default
exclude	Sequence[str]	List of attribute names to exclude from the changes. Defaults to fields that change every state update (e.g., last_updated, last_changed, context).	DEFAULT_EXCLUDE

Returns:

Type	Description
Callable[[RawStateChangeEvent], dict[str, dict[str, Any]]]	dict[str, dict[str, Any]]: A nested dict mapping names to tuples of (old_value, new_value).

Example return value

{
    "state": ("on", "off"),
    "attributes": {
        "rgb_color": ([255, 255, 0], [255, 255, 255]),
        "brightness": (200, 255)
    },
}

Source code in src/hassette/event_handling/accessors.py

def get_all_changes(
    exclude: Sequence[str] = DEFAULT_EXCLUDE,
) -> Callable[["RawStateChangeEvent"], dict[str, dict[str, Any]]]:
    """Get a dict of changed state and attribute values between old and new states.

    Args:
        exclude (Sequence[str], optional): List of attribute names to exclude from the changes. Defaults to fields
            that change every state update (e.g., `last_updated`, `last_changed`, `context`).

    Returns:
        dict[str, dict[str, Any]]: A nested dict mapping names to tuples of (old_value, new_value).

    Example return value:
        ```
        {
            "state": ("on", "off"),
            "attributes": {
                "rgb_color": ([255, 255, 0], [255, 255, 255]),
                "brightness": (200, 255)
            },
        }
        ```
    """

    def _inner(event: "RawStateChangeEvent") -> dict[str, Any]:
        old = cast("dict", event.payload.data.old_state or {})
        new = cast("dict", event.payload.data.new_state or {})

        changed_dict = _recursive_get_differences(old, new, exclude=exclude)

        return changed_dict

    return _inner

get_service(event: CallServiceEvent) -> str | FalseySentinel

Return the service name being called.

Source code in src/hassette/event_handling/accessors.py

def get_service(event: "CallServiceEvent") -> str | FalseySentinel:
    """Return the service name being called."""
    return get_path("payload.data.service")(event)

get_service_data(event: CallServiceEvent) -> dict[str, Any] | FalseySentinel

Return the service_data dict (or empty dict if missing).

Returns:

Type	Description
dict[str, Any] | FalseySentinel	dict[str, Any] | FalseySentinel: The service_data dict, or MISSING_VALUE if not present.

Source code in src/hassette/event_handling/accessors.py

def get_service_data(event: "CallServiceEvent") -> dict[str, Any] | FalseySentinel:
    """Return the service_data dict (or empty dict if missing).

    Returns:
        dict[str, Any] | FalseySentinel: The service_data dict, or MISSING_VALUE if not present.
    """
    return get_path("payload.data.service_data")(event)

get_service_data_key(key: str) -> Callable[[CallServiceEvent], Any]

Return an accessor that extracts a specific key from service_data.

Examples:

Basic equality against a literal

ValueIs(source=get_service_data_key("entity_id"), condition="light.living_room")

Callable condition (value must be an int > 200)

ValueIs(source=get_service_data_key("brightness"), condition=lambda v: isinstance(v, int) and v > 200)

Source code in src/hassette/event_handling/accessors.py

def get_service_data_key(key: str) -> "Callable[[CallServiceEvent], Any]":
    """Return an accessor that extracts a specific key from service_data.

    Examples:
    --------
    Basic equality against a literal

        ValueIs(source=get_service_data_key("entity_id"), condition="light.living_room")

    Callable condition (value must be an int > 200)

        ValueIs(source=get_service_data_key("brightness"), condition=lambda v: isinstance(v, int) and v > 200)

    """

    def _inner(event: "CallServiceEvent") -> Any:
        service_data = get_service_data(event)
        if service_data is MISSING_VALUE:
            return MISSING_VALUE

        if typing.TYPE_CHECKING:
            assert not isinstance(service_data, FalseySentinel)

        return service_data.get(key, MISSING_VALUE)

    return _inner