Skip to content

Dependency Injection

Instead of manually parsing event payloads, you declare what data you need using type annotations, and Hassette extracts and converts it for you. This is called dependency injection (DI).

Quick Example

from hassette import App, D, states


class LightMonitor(App):
    async def on_initialize(self):
        await self.bus.on_state_change(
            "light.bedroom",
            handler=self.on_light_change,
            name="bedroom_light",
        )

    async def on_light_change(self, new_state: D.StateNew[states.LightState], entity_id: D.EntityId):
        brightness = new_state.attributes.brightness
        self.logger.info("%s brightness: %s", entity_id, brightness)

In this example, new_state and entity_id are automatically extracted from the RawStateChangeEvent and injected into your handler based on their type annotations.

Three Event Handling Patterns

Hassette supports three patterns for handling events. The Writing Handlers page introduces all three starting from the simplest (raw events). Here they are ordered by what you'll use most in production code.

DI Extraction

Extract only the specific data you need:

from hassette import App, D, states


class MotionApp(App):
    async def on_motion(self, new_state: D.StateNew[states.BinarySensorState], entity_id: D.EntityId):
        friendly_name = new_state.attributes.friendly_name or entity_id
        self.logger.info("Motion detected: %s", friendly_name)

Use when: You want clean, focused handlers with minimal boilerplate.

Typed Event

Receive the full event with state objects converted to typed Pydantic models:

from hassette import App, D, states


class MotionApp(App):
    async def on_motion(
        self,
        event: D.TypedStateChangeEvent[states.BinarySensorState],
    ):
        entity_id = event.payload.data.entity_id
        new_state = event.payload.data.new_state
        if new_state:
            state_value = new_state.value
            self.logger.info("Motion: %s -> %s", entity_id, state_value)

Use when: You want type safety but need access to the full event structure (context, timestamps, metadata, etc.).

Raw Event (Untyped)

Receive the full event object with state data as untyped dictionaries:

from hassette import App
from hassette.events import RawStateChangeEvent


class MotionApp(App):
    async def on_motion(self, event: RawStateChangeEvent):
        entity_id = event.payload.data.entity_id
        new_state_dict = event.payload.data.new_state
        state_value = new_state_dict.get("state") if new_state_dict else None
        self.logger.info("Motion: %s -> %s", entity_id, state_value)

Use when: You need full control or are working with dynamic/unknown state structures.

Warning

While typed State models use value for the actual state value, raw state dictionaries are accessed via the "state" key, as this is the key used by Home Assistant in its event payloads, and this data is not modified by Hassette.

Available DI Annotations

All dependency injection annotations are available in the hassette.dependencies module (commonly imported as D).

State Object Extractors

Extract typed state objects from state change events:

Annotation Type Description
StateNew[T] T Extract new state, raises if missing
StateOld[T] T Extract old state, raises if missing
MaybeStateNew[T] T or None Extract new state, returns None if missing
MaybeStateOld[T] T or None Extract old state, returns None if missing
from hassette import App, D, states


class LightApp(App):
    async def on_light_change(self, new_state: D.StateNew[states.LightState], old_state: D.MaybeStateOld[states.LightState]):
        if old_state:
            brightness_changed = new_state.attributes.brightness != old_state.attributes.brightness
            if brightness_changed:
                self.logger.info(
                    "Brightness: %s -> %s",
                    old_state.attributes.brightness,
                    new_state.attributes.brightness,
                )

Note

In actual usage you should pass a condition to the changed parameter of on_state_change, which will handle this condition for you.

# `old` and `new` are the raw state strings (e.g. "on", "off", "unknown").
# This example only fires when the state was previously known and actually changed.
await self.bus.on_state_change(
    entity_id="light.office",
    handler=self.on_light_change,
    changed=lambda old, new: old is not None and new is not None and old != new,
    name="office_light_change",
)

Identity Extractors

Extract entity IDs and domains from events:

Annotation Type Description
EntityId str Extract entity ID, raises if missing
MaybeEntityId str or MISSING_VALUE (falsy sentinel) Extract entity ID, returns sentinel if missing
Domain str Extract domain, raises if missing
MaybeDomain str or MISSING_VALUE (falsy sentinel) Extract domain, returns sentinel if missing
from hassette import App, D


class LightApp(App):
    async def on_any_light(self, entity_id: D.EntityId, domain: D.Domain):
        self.logger.info("Light entity %s in domain %s changed", entity_id, domain)

Check with truthiness, not isinstance

MISSING_VALUE is a FalseySentinel — it is always falsy. Use if entity_id: to check for a present value rather than isinstance checks. FalseySentinel is not a public type and should not be used in type checks directly.

async def handler(entity_id: D.MaybeEntityId):
    if entity_id:
        # entity_id is a str here
        ...
    else:
        # entity_id is MISSING_VALUE (falsy)
        ...

Other Extractors

Annotation Type Description
EventData[T] T Extract typed data from a Bus.emit broadcast event
EventContext HassContext Extract Home Assistant event context
TypedStateChangeEvent[T] TypedStateChangeEvent[T] Convert raw event to typed event
from hassette import App, D, states


class LightApp(App):
    async def on_light_change(
        self,
        new_state: D.StateNew[states.LightState],
        context: D.EventContext,
    ):
        self.logger.info(
            "Light %s changed by user %s",
            new_state.entity_id,
            context.user_id or "system",
        )

EventData[T] extracts the typed payload from events sent via Bus.emit. The sender emits a plain dataclass; the receiver declares it as a parameter type:

from dataclasses import dataclass

from hassette import App, AppConfig, D


@dataclass(frozen=True, slots=True)
class SensorAlert:
    sensor_id: str
    reading: float


class AlertApp(App[AppConfig]):
    async def on_initialize(self) -> None:
        await self.bus.on(topic="sensor.alert", handler=self.on_alert, name="alert_handler")

    async def on_alert(self, alert: D.EventData[SensorAlert]) -> None:
        self.logger.warning("Sensor %s reading: %.1f", alert.sensor_id, alert.reading)

Union Type Support

DI extractors support Union types, allowing handlers to work with multiple state types:

from hassette import App, D, states


class SensorApp(App):
    async def on_sensor_change(self, new_state: D.StateNew[states.SensorState | states.BinarySensorState], entity_id: D.EntityId):
        # new_state is automatically converted to the correct type
        # based on the entity's domain
        if isinstance(new_state, states.SensorState):
            self.logger.info("Sensor %s: %s", entity_id, new_state.value)
        elif isinstance(new_state, states.BinarySensorState):
            self.logger.info("Binary sensor %s: %s", entity_id, new_state.value)

The StateRegistry determines the correct state class based on the entity's domain, and the DI system converts the raw state dictionary to the appropriate Pydantic model.

Combining Multiple Dependencies

You can extract multiple pieces of data in a single handler:

from hassette import App, D, states


class ClimateApp(App):
    async def on_climate_change(
        self,
        new_state: D.StateNew[states.ClimateState],
        old_state: D.MaybeStateOld[states.ClimateState],
        entity_id: D.EntityId,
        context: D.EventContext,
    ):
        old_temp = old_state.attributes.current_temperature if old_state else None
        new_temp = new_state.attributes.current_temperature

        if old_temp != new_temp:
            self.logger.info(
                "Climate %s temperature changed: %s -> %s (user: %s)",
                entity_id,
                old_temp,
                new_temp,
                context.user_id or "system",
            )

Mixing DI with Custom kwargs

Dependency injection works with custom keyword arguments passed when registering handlers:

from hassette import App, D, states


class TempApp(App):
    async def on_initialize(self):
        await self.bus.on_state_change(
            "sensor.temperature",
            handler=self.on_temp_change,
            kwargs={"threshold": 75.0, "message": "Temperature %s (%.1f°F) exceeds threshold %.1f°F"},
            name="temp_threshold_alert",
        )

    async def on_temp_change(self, new_state: D.StateNew[states.SensorState], entity_id: D.EntityId, threshold: float, message: str):
        temp = float(new_state.value) if new_state.value else 0.0
        if temp > threshold:
            self.logger.warning(message, entity_id, temp, threshold)

Advanced: Custom Extractors and Type Conversion

The rest of this page covers custom extractors and type conversion

If the built-in annotations above handle your use case, you can skip ahead to Handlers or Filtering.

Custom Extractors

You can create custom extractors using the Annotated type with either existing accessors from accessors or custom callables:

Using Built-in Accessors

from typing import Annotated

from hassette import A, App


class LightApp(App):
    async def on_light_change(
        self,
        brightness: Annotated[float | None, A.get_attr_new("brightness")],
        color_temp: Annotated[int | None, A.get_attr_new("color_temp")],
    ):
        self.logger.info("Brightness: %s, Color temp: %s", brightness, color_temp)

Writing Your Own Extractor

Any callable that accepts an event and returns a value can be used as an extractor:

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)

Advanced: Extractor + Converter Pattern

For more complex scenarios, you can use the AnnotationDetails class to combine extraction and type conversion:

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)

Automatic Type Conversion with TypeRegistry

Hassette's dependency injection system uses the TypeRegistry to automatically convert extracted values to match your type annotations. This works automatically with custom extractors.

How It Works

When you use a custom extractor with a type annotation, the DI system:

  1. Extracts the value using your extractor function
  2. Checks the type of the extracted value against your annotation
  3. Automatically converts if needed using the TypeRegistry
  4. Injects the converted value into your handler

This means you can write simple extractors that return raw values, and let TypeRegistry handle the type conversion:

from typing import Annotated

from hassette import A, App


class LightApp(App):
    async def on_light_change(
        self,
        # Extractor returns string "200" from HA
        # TypeRegistry automatically converts to int
        brightness: Annotated[int | None, A.get_attr_new("brightness")],
        # Extractor returns string "on" from HA
        # TypeRegistry automatically converts to bool
        is_on: Annotated[bool, A.get_state_value_new],
    ):
        if is_on and brightness and brightness > 200:
            self.logger.info("Light is very bright: %d", brightness)

Built-in Conversions

The TypeRegistry includes built-in conversions for common types:

  • Numeric types: strint, float, Decimal
  • Boolean: strbool (handles "on", "off", "true", "false", etc.)
  • DateTime types: strdatetime, date, time (stdlib), and whenever types

Examples:

from datetime import datetime
from decimal import Decimal
from typing import Annotated

from hassette import A, App


class SensorApp(App):
    async def on_sensor_change(
        self,
        # String "23.5" → float 23.5
        temperature: Annotated[float, A.get_attr_new("temperature")],
        # String "99" → int 99
        battery: Annotated[int | None, A.get_attr_new("battery_level")],
        # String "0.1234" → Decimal("0.1234") (high precision)
        precise_value: Annotated[Decimal | None, A.get_attr_new("value")],
        # ISO string → datetime object
        last_seen: Annotated[datetime | None, A.get_attr_new("last_seen")],
    ):
        self.logger.info(
            "Temp: %.1f°C, Battery: %d%%, Precise: %s, Last seen: %s",
            temperature,
            battery or 0,
            precise_value,
            last_seen,
        )

Custom Type Converters

You can register your own type converters for custom types:

from enum import StrEnum, auto
from typing import Annotated

from hassette import A, App, register_type_converter_fn


class Effect(StrEnum):
    BLINK = auto()
    BREATHE = auto()
    CANDLE = auto()
    CHANNEL_CHANGE = auto()
    COLORLOOP = auto()
    FINISH_EFFECT = auto()
    FIREPLACE = auto()
    OKAY = auto()
    STOP_EFFECT = auto()
    STOP_HUE_EFFECT = auto()


@register_type_converter_fn(error_message="'{value}' is not a valid Effect")
def str_to_effect(value: str) -> Effect:
    """Convert string to Effect enum.

    Types are inferred from the function signature.
    """
    return Effect(value.lower())


# Now you can use it in handlers


class LightEffectApp(App):
    async def on_light_effect_change(self, effect: Annotated[Effect, A.get_attr_new("effect")]):
        self.logger.info("Light effect: %r", effect)

When Conversion Happens

Type conversion is skipped if the returned value is already the correct type or is None.

If the value is not None and does not match the expected type, Hassette will attempt to convert it using the TypeRegistry. The TypeRegistry will first look for a registered converter for the (from_type, to_type) pair. If to_type is a tuple, it will iterate through each type in the tuple and use the first converter that succeeds.

If there is no registered converter for the (from_type, to_type) pair, Hassette will attempt to call to_type as a constructor with the value as the sole argument.

If type conversion fails, Hassette will raise a UnableToConvertValueError. For tuples, this will be raised only if all conversions fail.

Bypassing Automatic Conversion

If you want to handle conversion yourself, you can:

  1. Use Any type annotation to receive the raw value:
from typing import Annotated, Any

from hassette import A, App


class RawApp(App):
    async def handler(
        self,
        # No conversion - receive raw value
        raw_value: Annotated[Any, A.get_attr_new("brightness")],
    ):
        # Handle conversion yourself
        brightness = int(raw_value) if raw_value else None
        self.logger.info("Brightness: %s", brightness)
  1. Provide a custom converter in AnnotationDetails:
from typing import Annotated, Any

from hassette import A
from hassette.event_handling.dependencies import AnnotationDetails


def my_converter(value: Any, _: type) -> int | None:
    if value is None:
        return None
    return int(value) * 100


BrightnessPercent = Annotated[
    int,
    AnnotationDetails(
        extractor=A.get_attr_new("brightness"),
        converter=my_converter,
    ),
]

Error Handling

When type conversion fails, Hassette provides clear error messages:

from typing import Annotated

from hassette import A, App


class ErrorApp(App):
    async def handler(self, value: Annotated[int, A.get_attr_new("invalid_field")]):
        pass

    # Listener error (topic=hass.event.state_changed): Handler 'my_project.main.ErrorApp.handler' -
    # failed to convert parameter 'value' of type 'FalseySentinel' to type 'int': Unable to convert
    # <MISSING_VALUE> to <class 'int'>

Handler Signature Restrictions

DI handlers have some restrictions to ensure unambiguous parameter injection:

Handler Signature Rules

Handlers using DI cannot have:

  • Positional-only parameters (parameters before /)
  • Variadic positional arguments (*args)

These restrictions ensure that Hassette can reliably match parameters to extracted values.

Type Annotations Required

All parameters using dependency injection must have type annotations. Hassette uses these annotations to determine what to extract from events and how to convert the data.

How It Works

Under the hood, Hassette's DI system:

  1. Inspects handler signatures using Python's inspect module to find annotated parameters
  2. Extracts type information from Annotated types and recognizes special DI annotations
  3. Builds extractors for each parameter that know how to pull data from events
  4. Converts types using the StateRegistry for state objects, converting raw dictionaries to typed Pydantic models
  5. Injects values at call time, passing extracted and converted values as keyword arguments

The core implementation lives in:

See Also