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:
- Extracts the value using your extractor function
- Checks the type of the extracted value against your annotation
- Automatically converts if needed using the TypeRegistry
- 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:
str↔int,float,Decimal - Boolean:
str→bool(handles"on","off","true","false", etc.) - DateTime types:
str→datetime,date,time(stdlib), andwhenevertypes
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:
- Use
Anytype 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)
- 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:
- Inspects handler signatures using Python's
inspectmodule to find annotated parameters - Extracts type information from
Annotatedtypes and recognizes special DI annotations - Builds extractors for each parameter that know how to pull data from events
- Converts types using the StateRegistry for state objects, converting raw dictionaries to typed Pydantic models
- Injects values at call time, passing extracted and converted values as keyword arguments
The core implementation lives in:
extraction- Signature inspection and parameter extractiondependencies- Pre-defined DI annotationsaccessors- Low-level event data accessors
See Also
- Type Registry - automatic type conversion system
- State Registry - domain to state model mapping