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 two fields:

Field Type Required Purpose
extractor Callable[[T], Any] Yes Extracts the value from the event
converter Callable[[Any, Any], Any] \| None No Converts the extracted value to the declared type

Placing an AnnotationDetails instance inside Annotated[T, AnnotationDetails(...)] completes the setup. Hassette discovers AnnotationDetails in 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 (which has event.payload.data.new_state, event.payload.data.old_state, and event.payload.data.entity_id) and returns a string. 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.

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.event_handling.dependencies 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, covering types the built-in registry does not handle. See State Conversion for the full type registry.

See Also