Writing Event Handlers
A handler is an async method on an App that runs when an event matches a subscription.
Hassette supports four handler patterns: no parameters, extracted data via
dependency injection, raw events, and typed events. Choosing a Pattern summarizes when to use each.
Handler Patterns
No data needed
A no-parameter handler fires as a side effect. Hassette passes no event data. This pattern works with all subscription methods.
from hassette import App
class HeartbeatApp(App):
async def on_heartbeat(self) -> None:
self.logger.info("Heartbeat received")
Extracted data (recommended)
D (hassette.event_handling.dependencies) is a module of type annotations that tell Hassette what to extract from each event — similar to FastAPI's Depends(), but using type annotations instead of wrapper calls. The handler receives only the requested data, not the event object.
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)
D.StateNew[T] delivers the new state converted
to type T. D.EntityId delivers the entity ID
string.
The same pattern works with on_call_service. D.EntityId extracts the
entity the service call targeted.
from hassette import App, AppConfig, D
class LightAuditApp(App[AppConfig]):
async def on_initialize(self) -> None:
await self.bus.on_call_service(
"light",
handler=self.on_light_service,
name="light_service_audit",
)
async def on_light_service(
self,
entity_id: D.EntityId,
) -> None:
self.logger.info("Light service called on: %s", entity_id)
states is hassette.models.states, typed state classes for each Home Assistant domain. The Dependency Injection page covers the full annotation table, D.StateOld, D.EventContext, union types, and custom extractors.
Raw event
State change events arrive as
RawStateChangeEvent.
The state value lives at event.payload.data.new_state.get("state").
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_value = event.payload.data.new_state.get("state") if event.payload.data.new_state else None
self.logger.info("Motion: %s -> %s", entity_id, new_value)
Raw topic subscriptions via on() deliver Event[Any] instead. The handler
receives the full event object with event.topic and event.payload.
from typing import Any
from hassette import App, AppConfig
from hassette.events import Event
class TopicApp(App[AppConfig]):
async def on_initialize(self) -> None:
await self.bus.on(
topic="hass.event.automation_triggered",
handler=self.on_automation,
name="automation_triggered",
)
async def on_automation(
self, event: Event[Any]
) -> None:
self.logger.info("Topic: %s", event.topic)
Typed state event
D.TypedStateChangeEvent[T] converts
a raw state change event into a typed version with both old and new states as typed objects. D.StateNew[T] extracts just the new state; D.TypedStateChangeEvent[T] gives the full event — useful when comparing before/after values.
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
if event.payload.data.new_state:
new_value = event.payload.data.new_state.value
self.logger.info("Motion: %s -> %s", entity_id, new_value)
This pattern works only with on_state_change and on_attribute_change.
Service call handlers use the extracted data
pattern with D.EntityId instead.
Choosing a Pattern
D annotations are the default for most handlers. They deliver only the fields the handler needs. The signature stays readable, and Hassette handles parsing and type conversion.
Raw events deliver the full event structure, which suits event-forwarding or generic logging. Typed state events provide the same structure but with typed state objects instead of raw dicts.
No-parameter handlers work when the event itself does not matter. The subscription filters to the right entity and transition, so the handler just acts.
Cross-app Communication
Apps can broadcast data to other apps through custom topics.
Bus.emit(topic, data) publishes a payload. Other apps subscribe to the same
topic with on(). D.EventData[T] delivers the
payload pre-extracted and typed.
@dataclass(frozen=True)
class LightsSyncedData:
source: str
class SenderApp(App[AppConfig]):
async def on_initialize(self) -> None:
await self.bus.on_state_change(
"light.kitchen",
handler=self.on_kitchen_change,
name="kitchen_light",
)
async def on_kitchen_change(
self,
state: D.StateNew[states.LightState],
) -> None:
await self.bus.emit(
"lights_synced",
LightsSyncedData(source=self.instance_name),
)
class ReceiverApp(App[AppConfig]):
async def on_initialize(self) -> None:
await self.bus.on(
topic="lights_synced",
handler=self.on_lights_synced,
name="lights_synced_log",
)
async def on_lights_synced(
self,
data: D.EventData[LightsSyncedData],
) -> None:
self.logger.info("Synced by %s", data.source)
A frozen dataclass or Pydantic model works well for T — the type is passed in-process, not persisted, but keeping it immutable prevents accidental cross-app state mutation. Any type passed as data to emit() can be received via D.EventData[T]. self.instance_name is the app's instance identifier, set in hassette.toml.
See Also
- Subscription Methods: method reference, parameters, error handling, registration
- Dependency Injection: full
D.*annotation table, custom extractors - Filtering & Predicates: predicates, conditions,
where=usage