Apps Overview

Apps are the code you write to respond to events and control your home. Each app has its own behavior, configuration, and internal state.

Apps can be asynchronous (preferred) or synchronous. Sync apps are automatically run in threads to prevent blocking the event loop.

Structure

flowchart TD
    subgraph app["Your App"]
        A["App"]
    end

    subgraph resources["Resources"]
        direction LR
        Api
        Bus
        Scheduler
        States
        Cache
    end

    A --> Api & Bus & Scheduler & States & Cache

    style app fill:#e8f0ff,stroke:#6688cc
    style resources fill:#fff0e8,stroke:#cc8844

Hold "Ctrl" to enable pan & zoom

Defining an App

Every app is a Python class that inherits from App or AppSync.

example_app.py

from hassette import App, AppConfig, D, states


class ExampleApp(App[AppConfig]):
    async def on_initialize(self):
        self.logger.info("Hello from ExampleApp!")

        # Subscribe using the bus helper
        await self.bus.on_state_change(
            "light.living_room",
            handler=self.on_light_change,
            name="living_room_light",
        )

    async def on_light_change(
        self,
        new_state: D.StateNew[states.LightState],
    ):
        self.logger.info("Light changed to %s", new_state.value)

What's D.StateNew[states.LightState]?

That annotation is dependency injection — you declare what data you need, and Hassette extracts and types it from the event automatically. The Writing Handlers page covers how it works. For now, just notice the pattern.

Dates and Times

Hassette uses the whenever library for timezone-aware date/time handling instead of Python's stdlib datetime. Python's datetime has a mutable API and makes it easy to accidentally create "naive" (timezone-unaware) objects — a common source of bugs in time-sensitive automations. whenever is always timezone-aware and immutable, so incorrect comparisons between naive and aware times become type errors rather than silent failures. Every app provides self.now(), which returns a ZonedDateTime in your system timezone.

from whenever import TimeDelta, ZonedDateTime

last_seen: ZonedDateTime = self.now()
next_run = self.now().add(hours=2)  # 2 hours from now
elapsed: TimeDelta = self.now() - last_seen

You'll see ZonedDateTime in scheduler parameters, persistent storage examples, and custom state definitions. If you're familiar with datetime.datetime, the API is similar but always timezone-aware.

Core Capabilities

Each app receives pre-configured helpers:

self.api - Interact with Home Assistant.
self.bus - Subscribe to events.
self.scheduler - Schedule jobs.
self.states - Access entity states.
self.cache - Persistent disk-based storage.
self.logger - Dedicated logger instance.
self.app_config - Typed configuration.
self.task_bucket - Spawn background tasks and offload blocking work to a thread pool.

Common Use Cases

Reacting to Events

Subscribe to events using self.bus to react to changes in Home Assistant.

self.on_change_listener = await self.bus.on_state_change("light.kitchen", handler=self.on_change, name="kitchen_light")

Run Recurring Jobs

Use self.scheduler to schedule recurring tasks.

await self.scheduler.run_hourly(self.log_status)

Check Entity States

Use self.states to check the current state of entities.

current_state = self.states.light["light.kitchen"].value
self.logger.info("Current state: %s", current_state)

Call Services

Use self.api to call Home Assistant services.

await self.api.call_service("light", "turn_on", entity_id="light.kitchen")

Forgetting await on API calls

Every self.api.* method is a coroutine — it must be awaited. Writing self.api.call_service(...) without await returns a coroutine object and silently does nothing: no error is raised, no service is called, and no log message appears. If an API call seems to have no effect, check that you haven't dropped the await.

Persist Data Between Restarts

Use self.cache to store data that should survive app restarts.

# Load counter from cache, defaulting to 0
self.counter = self.cache.get("counter", 0)  # pyright: ignore[reportAttributeAccessIssue]

# Increment and save back
self.counter += 1
self.cache["counter"] = self.counter

Run Background Tasks and Blocking Code

Use self.task_bucket to spawn fire-and-forget coroutines or offload blocking calls to a thread pool. All tracked tasks are cancelled automatically on shutdown.

# Fire off a background coroutine — the bucket tracks and cancels it on shutdown
self.task_bucket.spawn(self.poll_sensor(), name="poll_sensor")

See the Task Bucket page for the full API: spawn(), run_in_thread(), make_async_adapter(), and cross-thread communication.

Restricting to a Single App During Development

The @only_app decorator prevents multiple instances of the same app class from running. Apply it during development or testing when you want to isolate one app without editing your configuration files:

from hassette import App, AppConfig, only_app
from pydantic_settings import SettingsConfigDict


class MyConfig(AppConfig):
    model_config = SettingsConfigDict(env_prefix="my_")


@only_app
class MyApp(App[MyConfig]):
    ...

If more than one class in your project is decorated with @only_app, Hassette raises an error at startup. Remove the decorator before deploying.

Broadcasting Events Between Apps

Bus.emit broadcasts an event to all apps subscribed to a given topic. The event stays in-process — it never reaches Home Assistant. All apps that called self.bus.on(topic=...) for that topic receive it.

This is the on/emit symmetry: subscribe with self.bus.on, broadcast with self.bus.emit. Both live on the same Bus instance.

class LightManagerApp(App[AppConfig]):
    async def on_initialize(self) -> None:
        await self.bus.on_state_change(
            "light.kitchen",
            handler=self.on_kitchen_change,
            name="kitchen_sync",
        )

    async def on_kitchen_change(self, state: D.StateNew[states.LightState]) -> None:
        await self.bus.emit("lights_synced", LightsSyncedData(source=self.instance_name))

The receiving app subscribes to the same topic and extracts the typed data via D.EventData[T] — a dependency injection annotation that Hassette resolves from the event envelope automatically.

class LoggerApp(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:
        if data.source == self.instance_name:
            return
        self.logger.info("Lights synced by %s", data.source)

Broadcast is local and ephemeral — events are not persisted across restarts and do not leave the framework process.

Self-delivery

An app that both subscribes to and emits on the same topic receives its own event. To filter self-emitted events, include a source field on the emitted dataclass (as LightsSyncedData does above) and guard in the handler: if data.source == self.instance_name: return.

Synchronous Apps

AppSync — for blocking code

AppSync is a subclass of App for automations that must call blocking (non-async) libraries. Instead of overriding on_initialize and on_shutdown, you override their _sync-suffixed counterparts (on_initialize_sync, on_shutdown_sync, etc.). Hassette runs these methods in a thread pool so they do not block the event loop.

The bus, scheduler, and API are async. From a _sync hook, reach their synchronous facades through .sync — self.bus.sync, self.scheduler.sync, and self.api.sync. Each facade method blocks until the underlying async call completes. Calling one from inside the event loop raises RuntimeError instead of deadlocking.

from hassette import AppSync

class MyApp(AppSync[MyConfig]):
    def on_initialize_sync(self) -> None:
        # registration runs through the .sync facades from sync code
        self.bus.sync.on_state_change("light.kitchen", handler=self.on_change, name="kitchen")
        self.scheduler.sync.run_in(self.cleanup, 60, name="cleanup")
        self.api.sync.call_service("light", "turn_on", target={"entity_id": "light.kitchen"})

    def on_change(self, event) -> None:
        ...

    def cleanup(self) -> None:
        ...

    def on_shutdown_sync(self) -> None:
        ...

Prefer async App whenever possible. Use AppSync only when a third-party library provides no async interface and wrapping it yourself is impractical.

Next Steps

Lifecycle: Understand on_initialize and on_shutdown.
Configuration: Learn how to use typed configuration and secrets.