Skip to content

Custom Extractors

The built-in D.* annotations cover state values, entity IDs, domains, event data, and event context. Custom extractors handle everything else: a specific key from service_data, a nested attribute, or a value computed from multiple event fields.

Accessors (A)

A (from hassette import A) provides accessor functions that target non-standard event fields. Accessors are the simplest form of custom extraction. They work directly as Annotated type metadata, with no additional wrapping.

A.get_attr_new("brightness") returns a callable that extracts brightness from the new state's attributes. A.get_service_data_key("entity_id") extracts a key from service_data (the dict of parameters passed to a service call). A.get_path("payload.data.new_state.attributes.geolocation.locality") traverses a dotted path. It returns MISSING_VALUE — a falsy sentinel — if any segment is absent.

from hassette import A, App, AppConfig, P
from hassette.events import CallServiceEvent, RawStateChangeEvent


class MyApp(App[AppConfig]):
    async def on_initialize(self) -> None:
        # Only handle turn_on calls targeting a specific entity
        entity_match = P.ValueIs(source=A.get_service_data_key("entity_id"), condition="light.living_room")
        await self.bus.on_call_service(domain="light", service="turn_on", handler=self.on_living_room_on, where=entity_match, name="living_room_turn_on")

        # Check a nested attribute value using a dotted path
        city_match = 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=self.on_location_change, changed_to=city_match, name="device_location")

    async def on_living_room_on(self, event: CallServiceEvent) -> None: ...
    async def on_location_change(self, event: RawStateChangeEvent) -> None: ...

Accessors also compose with predicates. P.ValueIs(source=A.get_service_data_key("entity_id"), condition="light.living_room") filters a service call subscription to a specific target entity without any handler logic. The full predicate reference is in Filtering.

Writing an Extractor

A custom extractor is a plain callable that receives the raw event and returns a value. AnnotationDetails wraps that callable and registers it with the DI system.

AnnotationDetails is a frozen dataclass with three fields:

Field Type Required Purpose
extractor Callable[[T], Any] Yes Extracts the value from the source object
converter Callable[[Any, Any], Any] \| None No Converts the extracted value to the declared type
source_type type[T] \| None No Overrides the default source type for this extractor

Annotated[T, AnnotationDetails(...)] completes the setup. Hassette discovers the AnnotationDetails instance in the Annotated metadata automatically at registration time — no explicit registration step needed.

from typing import Annotated

from hassette import App
from hassette.events import RawStateChangeEvent


def get_friendly_name(event: RawStateChangeEvent) -> str:
    """Extract friendly_name from new state attributes."""
    new_state = event.payload.data.new_state
    if new_state and "attributes" in new_state:
        return new_state["attributes"].get("friendly_name", "Unknown")
    return "Unknown"


class MyCustomExtractorApp(App):
    async def on_state_change(
        self,
        name: Annotated[str, get_friendly_name],
    ):
        self.logger.info("Changed: %s", name)

get_friendly_name receives the raw RawStateChangeEvent and returns a string. The event exposes event.payload.data.new_state, event.payload.data.old_state, and event.payload.data.entity_id. The Annotated[str, get_friendly_name] annotation tells the DI system to call that function for name on each invocation.

A plain callable in the Annotated metadata position is the simplest form — Hassette wraps it in AnnotationDetails automatically. The explicit AnnotationDetails form above is needed only when adding a type converter or overriding the default source_type.

Adding Type Conversion

AnnotationDetails.converter accepts a function with the signature (value: Any, to_type: type) -> Any. The DI system calls it after extraction to convert the raw value to the declared type.

from datetime import datetime
from typing import Annotated

from hassette import App
from hassette.di import AnnotationDetails
from hassette.events import RawStateChangeEvent


def extract_timestamp(event: RawStateChangeEvent) -> str | None:
    """Extract last_changed timestamp from new state."""
    new_state = event.payload.data.new_state
    return new_state.get("last_changed", None) if new_state else None


def convert_to_datetime(value: str, _to_type: type) -> datetime:
    """Convert ISO string to datetime."""
    return datetime.fromisoformat(value.replace("Z", "+00:00"))


LastChanged = Annotated[
    datetime,
    AnnotationDetails(extractor=extract_timestamp, converter=convert_to_datetime),
]


class TimestampApp(App):
    async def on_state_change(
        self,
        changed_at: LastChanged,
    ):
        self.logger.info("State changed at: %s", changed_at)

extract_timestamp returns an ISO string. convert_to_datetime converts that string to a datetime. The LastChanged type alias bundles both into a reusable annotation. Any handler parameter typed as LastChanged receives a datetime with no inline parsing.

Hassette converts standard scalar types (int, float, bool, str) automatically — no converter needed for those. AnnotationDetails.converter handles conversions specific to a single extractor — types the built-in registry doesn't cover. See State Conversion for the full type registry.

See Also