App

App base classes and configuration for building Home Assistant automations.

This module provides clean access to the app framework for creating both async and sync applications with typed configuration.

BlockingIOBehavior

Bases: StrEnum

Controls what happens when blocking I/O is detected on the shared event loop.

Source code in src/hassette/types/enums.py

class BlockingIOBehavior(StrEnum):
    """Controls what happens when blocking I/O is detected on the shared event loop."""

    IGNORE = auto()
    """Suppress detection entirely — no warning AND no ``blocking_events`` row is written.
    This is the deliberate exception to the persist-before-warn rule: ``IGNORE`` means
    "I know this app blocks; treat it as if detection never fired," so there is no audit
    trail for ignored events."""

    WARN = auto()
    """Emit a ``HassetteBlockingIOWarning`` (default). Integrates with ``-W error``."""

    ERROR = auto()
    """Emit ``HassetteBlockingIOWarning`` in a form that ``filterwarnings("error")`` escalates
    to a raised exception. Under normal filters, behaves identically to ``WARN`` — the framework
    does not raise on its own; escalation requires the caller to install the error filter
    (``-W error``, a pytest ``filterwarnings`` marker, or ``warnings.filterwarnings`` at startup).

    The effect of escalation differs by detection tier:

    - **Tier 2 (call-site interception)** detects *before* the blocking primitive runs, so an
      escalated warning raises at the call site and prevents the blocking call from executing.
    - **Tier 1 (loop watchdog)** is warn-after: the stall has already completed when the warning
      fires on the off-loop daemon thread. The daemon suppresses the escalation to stay alive, so
      ``ERROR`` cannot retroactively interrupt a Tier 1 stall — it only records and warns."""

IGNORE = auto() class-attribute instance-attribute

Suppress detection entirely — no warning AND no blocking_events row is written. This is the deliberate exception to the persist-before-warn rule: IGNORE means "I know this app blocks; treat it as if detection never fired," so there is no audit trail for ignored events.

WARN = auto() class-attribute instance-attribute

Emit a HassetteBlockingIOWarning (default). Integrates with -W error.

ERROR = auto() class-attribute instance-attribute

Emit HassetteBlockingIOWarning in a form that filterwarnings("error") escalates to a raised exception. Under normal filters, behaves identically to WARN — the framework does not raise on its own; escalation requires the caller to install the error filter (-W error, a pytest filterwarnings marker, or warnings.filterwarnings at startup).

The effect of escalation differs by detection tier:

• Tier 2 (call-site interception) detects before the blocking primitive runs, so an escalated warning raises at the call site and prevents the blocking call from executing.
• Tier 1 (loop watchdog) is warn-after: the stall has already completed when the warning fires on the off-loop daemon thread. The daemon suppresses the escalation to stay alive, so ERROR cannot retroactively interrupt a Tier 1 stall — it only records and warns.

ForgottenAwaitBehavior

Bases: StrEnum

Controls what happens when a protected method is called without await.

Source code in src/hassette/types/enums.py

class ForgottenAwaitBehavior(StrEnum):
    """Controls what happens when a protected method is called without ``await``."""

    IGNORE = auto()
    """Suppress the warning entirely — the forgotten await is silently ignored."""

    WARN = auto()
    """Emit a ``HassetteForgottenAwaitWarning`` (default). Integrates with ``-W error``."""

    ERROR = auto()
    """Emit ``HassetteForgottenAwaitWarning`` in a form that ``filterwarnings("error")`` escalates
    to a raised exception. Under normal filters, behaves identically to ``WARN``."""

IGNORE = auto() class-attribute instance-attribute

Suppress the warning entirely — the forgotten await is silently ignored.

WARN = auto() class-attribute instance-attribute

Emit a HassetteForgottenAwaitWarning (default). Integrates with -W error.

ERROR = auto() class-attribute instance-attribute

Emit HassetteForgottenAwaitWarning in a form that filterwarnings("error") escalates to a raised exception. Under normal filters, behaves identically to WARN.

App

Bases: Generic[AppConfigT], Resource

Base class for applications in the Hassette framework.

This class provides a structure for applications, allowing them to be initialized and managed within the Hassette ecosystem. Lifecycle will generally be managed for you via the service status events, which send an event to the Bus and set the status attribute, based on the app's lifecycle.

Source code in src/hassette/app/app.py

class App(Generic[AppConfigT], Resource, metaclass=FinalMeta):
    """Base class for applications in the Hassette framework.

    This class provides a structure for applications, allowing them to be initialized and managed
    within the Hassette ecosystem. Lifecycle will generally be managed for you via the service status events,
    which send an event to the Bus and set the `status` attribute, based on the app's lifecycle.
    """

    _only_app: ClassVar[bool] = False
    """If True, only this app will be run. Only one app can be marked as only."""

    _import_exception: ClassVar[Exception | None] = None
    """Exception raised during import, if any. This prevents having all apps in a module fail due to one exception."""

    default_cache_ttl: ClassVar[int | None] = None
    """Default TTL (seconds) for this app's cache entries. Falls back to
    ``hassette.config.default_cache_ttl`` when unset, and to no expiration when both are unset."""

    role: ClassVar[ResourceRole] = ResourceRole.APP
    """Role of the resource, e.g. 'App', 'Service', etc."""

    source_tier: ClassVar[SourceTier] = "app"

    app_manifest: AppManifest | None
    """Manifest describing this app instance, set at construction from its config section.

    Stored per instance, not on the class: two config sections can point at the same
    App subclass, so each instance carries its own manifest rather than reading a
    class-shared one — otherwise the section loaded last would overwrite display_name,
    enabled, and auto_loaded for every instance. Mirrors how app_key is stored per
    instance. None only when an App is constructed directly without a manifest; the
    factory and the test harness both pass one.
    """

    app_config_cls: ClassVar[type[AppConfig]]
    """Config class to use for instances of the created app. Configuration from hassette.toml or
    other sources will be validated by this class."""

    logger: Logger
    """Logger for the instance."""

    api: "Api"
    """API instance for interacting with Home Assistant."""

    scheduler: "Scheduler"
    """Scheduler instance for scheduled tasks owned by this app."""

    bus: "Bus"
    """Event bus instance for event handlers owned by this app."""

    states: "StateManager"
    """States manager instance for accessing Home Assistant states."""

    app_config: AppConfigT
    """Configuration for this app instance."""

    index: int
    """Index of this app instance, used for unique naming."""

    cache: "CacheProtocol"
    """Cache instance for storing arbitrary data, scoped to this app instance."""

    def __init__(
        self,
        hassette: "Hassette",
        *,
        app_config: AppConfigT,
        index: int,
        app_key: str,
        app_manifest: AppManifest | None = None,
        api_factory: type[Resource] | None = None,
        parent: Resource | None = None,
        cache: CacheProtocol | None = None,
    ) -> None:
        # app_config and index must be set before super().__init__ because
        # unique_name (used by the logger) depends on app_config
        self.app_config = app_config
        self.index = index
        self._app_key = app_key
        self.app_manifest = app_manifest
        super().__init__(hassette, parent=parent)
        self.api = cast("Api", self.add_child(api_factory or Api))
        self.scheduler = self.add_child(Scheduler)
        self.bus = self.add_child(Bus, priority=0)
        self.states = self.add_child(StateManager)

        if cache is not None:
            self.cache = cache
        else:
            db_path = hassette.config.data_dir / self.cache_key / "cache" / "cache.db"
            default_ttl = (
                self.default_cache_ttl if self.default_cache_ttl is not None else hassette.config.default_cache_ttl
            )
            self.cache = AsyncCache(db_path, default_ttl)

    def __dir__(self) -> list[str]:
        return sorted(_APP_PUBLIC_API)

    @property
    def unique_name(self) -> str:
        """Unique name for the app instance, used for logging and ownership of resources."""
        if self.app_config.instance_name.startswith(self.class_name):
            return self.app_config.instance_name
        return f"{self.class_name}.{self.app_config.instance_name}"

    @property
    def config_log_level(self) -> LOG_LEVEL_TYPE:
        """Return the log level from the config for this resource."""
        try:
            return self.app_config.log_level
        except AttributeError:
            return self.hassette.config.logging.apps

    @property
    def app_key(self) -> str:
        """Key for this app in the hassette.toml configuration.

        Set per instance at construction. Two config sections can point at the same
        App subclass, so each instance stores its own key — reading it from a
        class-shared attribute would let the last section loaded clobber it for all.
        """
        return self._app_key

    @property
    def instance_name(self) -> str:
        """Name for the instance of the app. Used for logging and ownership of resources."""
        return self.app_config.instance_name

    @property
    def cache_key(self) -> str:
        """Key used to scope this app instance's cache directory.

        Returns the manifest's explicit ``cache_key`` when set; otherwise defaults to
        ``f"{app_key}/{index}"`` so each app instance gets its own cache directory.
        """
        if self.app_manifest and self.app_manifest.cache_key:
            return self.app_manifest.cache_key
        return f"{self.app_key}/{self.index}"

    def now(self) -> ZonedDateTime:
        """Return the current date and time."""
        return date_utils.now()

    async def before_initialize(self) -> None:
        """Optional: prepare to accept new work, allocate sockets, queues, temp files, etc.

        Initializes the cache (opens connections, creates schema, checks integrity) when
        it is a real ``AsyncCache``. When a cache was injected via the constructor (e.g. a
        ``DummyCache`` in tests), initialization is skipped -- ``DummyCache.initialize()``
        is a no-op, but the guard avoids the call entirely for clarity.
        """
        if isinstance(self.cache, AsyncCache):
            await self.cache.initialize()

    @final
    async def cleanup(self, timeout: float | None = None) -> None:
        """Cleanup resources owned by the instance.

        This method is called during shutdown to cancel tasks and close caches.
        Child cleanup (Bus, Scheduler, etc.) is handled by _finalize_shutdown() propagation,
        not by this method.
        """
        timeout = timeout or self.hassette.config.lifecycle.app_shutdown_timeout_seconds
        await super().cleanup(timeout=timeout)
        try:
            await self.cache.close()
        except Exception as exc:
            self.logger.exception("Error closing cache: %s %s", type(exc).__name__, exc)

default_cache_ttl: int | None = None class-attribute

Default TTL (seconds) for this app's cache entries. Falls back to hassette.config.default_cache_ttl when unset, and to no expiration when both are unset.

role: ResourceRole = ResourceRole.APP class-attribute

Role of the resource, e.g. 'App', 'Service', etc.

app_manifest: AppManifest | None = app_manifest instance-attribute

Manifest describing this app instance, set at construction from its config section.

Stored per instance, not on the class: two config sections can point at the same App subclass, so each instance carries its own manifest rather than reading a class-shared one — otherwise the section loaded last would overwrite display_name, enabled, and auto_loaded for every instance. Mirrors how app_key is stored per instance. None only when an App is constructed directly without a manifest; the factory and the test harness both pass one.

app_config_cls: type[AppConfig] class-attribute

Config class to use for instances of the created app. Configuration from hassette.toml or other sources will be validated by this class.

logger: Logger instance-attribute

Logger for the instance.

api: Api = cast('Api', self.add_child(api_factory or Api)) instance-attribute

API instance for interacting with Home Assistant.

scheduler: Scheduler = self.add_child(Scheduler) instance-attribute

Scheduler instance for scheduled tasks owned by this app.

bus: Bus = self.add_child(Bus, priority=0) instance-attribute

Event bus instance for event handlers owned by this app.

states: StateManager = self.add_child(StateManager) instance-attribute

States manager instance for accessing Home Assistant states.

app_config: AppConfigT = app_config instance-attribute

Configuration for this app instance.

index: int = index instance-attribute

Index of this app instance, used for unique naming.

cache: CacheProtocol instance-attribute

Cache instance for storing arbitrary data, scoped to this app instance.

unique_name: str property

Unique name for the app instance, used for logging and ownership of resources.

config_log_level: LOG_LEVEL_TYPE property

Return the log level from the config for this resource.

app_key: str property

Key for this app in the hassette.toml configuration.

Set per instance at construction. Two config sections can point at the same App subclass, so each instance stores its own key — reading it from a class-shared attribute would let the last section loaded clobber it for all.

instance_name: str property

Name for the instance of the app. Used for logging and ownership of resources.

cache_key: str property

Key used to scope this app instance's cache directory.

Returns the manifest's explicit cache_key when set; otherwise defaults to f"{app_key}/{index}" so each app instance gets its own cache directory.

now() -> ZonedDateTime

Return the current date and time.

Source code in src/hassette/app/app.py

def now(self) -> ZonedDateTime:
    """Return the current date and time."""
    return date_utils.now()

before_initialize() -> None async

Optional: prepare to accept new work, allocate sockets, queues, temp files, etc.

Initializes the cache (opens connections, creates schema, checks integrity) when it is a real AsyncCache. When a cache was injected via the constructor (e.g. a DummyCache in tests), initialization is skipped -- DummyCache.initialize() is a no-op, but the guard avoids the call entirely for clarity.

Source code in src/hassette/app/app.py

async def before_initialize(self) -> None:
    """Optional: prepare to accept new work, allocate sockets, queues, temp files, etc.

    Initializes the cache (opens connections, creates schema, checks integrity) when
    it is a real ``AsyncCache``. When a cache was injected via the constructor (e.g. a
    ``DummyCache`` in tests), initialization is skipped -- ``DummyCache.initialize()``
    is a no-op, but the guard avoids the call entirely for clarity.
    """
    if isinstance(self.cache, AsyncCache):
        await self.cache.initialize()

cleanup(timeout: float | None = None) -> None async

Cleanup resources owned by the instance.

This method is called during shutdown to cancel tasks and close caches. Child cleanup (Bus, Scheduler, etc.) is handled by _finalize_shutdown() propagation, not by this method.

Source code in src/hassette/app/app.py

@final
async def cleanup(self, timeout: float | None = None) -> None:
    """Cleanup resources owned by the instance.

    This method is called during shutdown to cancel tasks and close caches.
    Child cleanup (Bus, Scheduler, etc.) is handled by _finalize_shutdown() propagation,
    not by this method.
    """
    timeout = timeout or self.hassette.config.lifecycle.app_shutdown_timeout_seconds
    await super().cleanup(timeout=timeout)
    try:
        await self.cache.close()
    except Exception as exc:
        self.logger.exception("Error closing cache: %s %s", type(exc).__name__, exc)

AppSync

Bases: App[AppConfigT]

Synchronous adapter for App.

Source code in src/hassette/app/app.py

class AppSync(App[AppConfigT]):
    """Synchronous adapter for App."""

    def __dir__(self) -> list[str]:
        return sorted(_APP_PUBLIC_API | _APPSYNC_HOOKS)

    @final
    async def before_shutdown(self) -> None:
        """Optional: stop accepting new work, signal loops to wind down, etc."""
        await self.task_bucket.run_in_thread(self.before_shutdown_sync)

    @final
    async def on_shutdown(self) -> None:
        """Primary hook: release your own stuff (sockets, queues, temp files…)."""
        await self.task_bucket.run_in_thread(self.on_shutdown_sync)

    @final
    async def after_shutdown(self) -> None:
        """Optional: last-chance actions after on_shutdown, before cleanup/STOPPED."""
        await self.task_bucket.run_in_thread(self.after_shutdown_sync)

    @final
    async def before_initialize(self) -> None:
        """Optional: prepare to accept new work, allocate sockets, queues, temp files, etc."""
        await super().before_initialize()
        await self.task_bucket.run_in_thread(self.before_initialize_sync)

    @final
    async def on_initialize(self) -> None:
        """Primary hook: perform your own initialization (sockets, queues, temp files…)."""
        await self.task_bucket.run_in_thread(self.on_initialize_sync)

    @final
    async def after_initialize(self) -> None:
        """Optional: finalize initialization, signal readiness, etc."""
        await self.task_bucket.run_in_thread(self.after_initialize_sync)

    def before_shutdown_sync(self) -> None:
        """Optional: stop accepting new work, signal loops to wind down, etc."""
        pass

    def on_shutdown_sync(self) -> None:
        """Primary hook: release your own stuff (sockets, queues, temp files…)."""
        pass

    def after_shutdown_sync(self) -> None:
        """Optional: last-chance actions after on_shutdown, before cleanup/STOPPED."""
        pass

    def before_initialize_sync(self) -> None:
        """Optional: prepare to accept new work, allocate sockets, queues, temp files, etc."""
        pass

    def on_initialize_sync(self) -> None:
        """Primary hook: perform your own initialization (sockets, queues, temp files…)."""
        pass

    def after_initialize_sync(self) -> None:
        """Optional: finalize initialization, signal readiness, etc."""
        pass

    @final
    def initialize_sync(self) -> None:
        """Use on_initialize_sync instead."""
        raise NotImplementedError("Use on_initialize_sync instead.")

    @final
    def shutdown_sync(self) -> None:
        """Use on_shutdown_sync instead."""
        raise NotImplementedError("Use on_shutdown_sync instead.")

before_shutdown() -> None async

Optional: stop accepting new work, signal loops to wind down, etc.

Source code in src/hassette/app/app.py

@final
async def before_shutdown(self) -> None:
    """Optional: stop accepting new work, signal loops to wind down, etc."""
    await self.task_bucket.run_in_thread(self.before_shutdown_sync)

on_shutdown() -> None async

Primary hook: release your own stuff (sockets, queues, temp files…).

Source code in src/hassette/app/app.py

@final
async def on_shutdown(self) -> None:
    """Primary hook: release your own stuff (sockets, queues, temp files…)."""
    await self.task_bucket.run_in_thread(self.on_shutdown_sync)

after_shutdown() -> None async

Optional: last-chance actions after on_shutdown, before cleanup/STOPPED.

Source code in src/hassette/app/app.py

@final
async def after_shutdown(self) -> None:
    """Optional: last-chance actions after on_shutdown, before cleanup/STOPPED."""
    await self.task_bucket.run_in_thread(self.after_shutdown_sync)

before_initialize() -> None async

Optional: prepare to accept new work, allocate sockets, queues, temp files, etc.

Source code in src/hassette/app/app.py

@final
async def before_initialize(self) -> None:
    """Optional: prepare to accept new work, allocate sockets, queues, temp files, etc."""
    await super().before_initialize()
    await self.task_bucket.run_in_thread(self.before_initialize_sync)

on_initialize() -> None async

Primary hook: perform your own initialization (sockets, queues, temp files…).

Source code in src/hassette/app/app.py

@final
async def on_initialize(self) -> None:
    """Primary hook: perform your own initialization (sockets, queues, temp files…)."""
    await self.task_bucket.run_in_thread(self.on_initialize_sync)

after_initialize() -> None async

Optional: finalize initialization, signal readiness, etc.

Source code in src/hassette/app/app.py

@final
async def after_initialize(self) -> None:
    """Optional: finalize initialization, signal readiness, etc."""
    await self.task_bucket.run_in_thread(self.after_initialize_sync)

before_shutdown_sync() -> None

Optional: stop accepting new work, signal loops to wind down, etc.

Source code in src/hassette/app/app.py

def before_shutdown_sync(self) -> None:
    """Optional: stop accepting new work, signal loops to wind down, etc."""
    pass

on_shutdown_sync() -> None

Primary hook: release your own stuff (sockets, queues, temp files…).

Source code in src/hassette/app/app.py

def on_shutdown_sync(self) -> None:
    """Primary hook: release your own stuff (sockets, queues, temp files…)."""
    pass

after_shutdown_sync() -> None

Optional: last-chance actions after on_shutdown, before cleanup/STOPPED.

Source code in src/hassette/app/app.py

def after_shutdown_sync(self) -> None:
    """Optional: last-chance actions after on_shutdown, before cleanup/STOPPED."""
    pass

before_initialize_sync() -> None

Optional: prepare to accept new work, allocate sockets, queues, temp files, etc.

Source code in src/hassette/app/app.py

def before_initialize_sync(self) -> None:
    """Optional: prepare to accept new work, allocate sockets, queues, temp files, etc."""
    pass

on_initialize_sync() -> None

Primary hook: perform your own initialization (sockets, queues, temp files…).

Source code in src/hassette/app/app.py

def on_initialize_sync(self) -> None:
    """Primary hook: perform your own initialization (sockets, queues, temp files…)."""
    pass

after_initialize_sync() -> None

Optional: finalize initialization, signal readiness, etc.

Source code in src/hassette/app/app.py

def after_initialize_sync(self) -> None:
    """Optional: finalize initialization, signal readiness, etc."""
    pass

initialize_sync() -> None

Use on_initialize_sync instead.

Source code in src/hassette/app/app.py

@final
def initialize_sync(self) -> None:
    """Use on_initialize_sync instead."""
    raise NotImplementedError("Use on_initialize_sync instead.")

shutdown_sync() -> None

Use on_shutdown_sync instead.

Source code in src/hassette/app/app.py

@final
def shutdown_sync(self) -> None:
    """Use on_shutdown_sync instead."""
    raise NotImplementedError("Use on_shutdown_sync instead.")

AppConfig

Bases: BaseSettings

Base configuration class for applications in the Hassette framework.

This default class allows all extras, so arbitrary additional configuration data can be passed without needing to define a custom subclass, at the cost of type safety.

Fields can be set on subclasses and extra can be overridden by assigning a new value to model_config.

Source code in src/hassette/app/app_config.py

class AppConfig(BaseSettings):
    """Base configuration class for applications in the Hassette framework.

    This default class allows all extras, so arbitrary additional configuration data
    can be passed without needing to define a custom subclass, at the cost of type safety.

    Fields can be set on subclasses and extra can be overridden by assigning a new value to `model_config`.
    """

    model_config = SettingsConfigDict(
        extra="allow",
        arbitrary_types_allowed=True,
        env_file=ENV_FILE_LOCATIONS,
        env_ignore_empty=True,
        use_attribute_docstrings=True,
    )

    instance_name: str = ""
    """Name for the instance of the app."""

    log_level: LOG_LEVEL_TYPE = Field(default_factory=log_level_default_factory)
    """Log level for the app instance. Defaults to INFO if not provided."""

    app_key: str = ""
    """Configuration-level app key. Reserved: '__hassette__' and '__hassette__.*' prefixes are rejected."""

    forgotten_await_behavior: ForgottenAwaitBehavior | None = None
    """Per-app control for forgotten-await detection behavior.

    When ``None`` (default), the global ``HassetteConfig.forgotten_await_behavior`` is used,
    which itself defaults to ``"warn"``.  Set to ``"ignore"`` to suppress warnings for this app,
    or ``"error"`` to treat forgotten awaits as errors (escalated by ``filterwarnings("error")``)."""

    blocking_io_behavior: BlockingIOBehavior | None = None
    """Per-app control for blocking-IO detection behavior.

    When ``None`` (default), the global ``HassetteConfig.blocking_io.behavior`` is used,
    which itself defaults to ``"warn"``.  Set to ``"ignore"`` to suppress detection for this app,
    or ``"error"`` to escalate via ``filterwarnings("error")``."""

    @field_validator("app_key")
    @classmethod
    def _reject_hassette_sentinel(cls, v: str) -> str:
        if is_framework_key(v):
            raise ValueError(
                f"'{v}' is a reserved app_key used by the framework internally "
                f"(reserved prefix: '{FRAMEWORK_APP_KEY_PREFIX}'). "
                "Choose a different app_key for your application."
            )
        return v

instance_name: str = '' class-attribute instance-attribute

Name for the instance of the app.

log_level: LOG_LEVEL_TYPE = Field(default_factory=log_level_default_factory) class-attribute instance-attribute

Log level for the app instance. Defaults to INFO if not provided.

app_key: str = '' class-attribute instance-attribute

Configuration-level app key. Reserved: 'hassette' and 'hassette.*' prefixes are rejected.

forgotten_await_behavior: ForgottenAwaitBehavior | None = None class-attribute instance-attribute

Per-app control for forgotten-await detection behavior.

When None (default), the global HassetteConfig.forgotten_await_behavior is used, which itself defaults to "warn". Set to "ignore" to suppress warnings for this app, or "error" to treat forgotten awaits as errors (escalated by filterwarnings("error")).

blocking_io_behavior: BlockingIOBehavior | None = None class-attribute instance-attribute

Per-app control for blocking-IO detection behavior.

When None (default), the global HassetteConfig.blocking_io.behavior is used, which itself defaults to "warn". Set to "ignore" to suppress detection for this app, or "error" to escalate via filterwarnings("error").

only_app(app_cls: type[AppT]) -> type[AppT]

Decorator to mark an app class as the only one to run. If more than one app is marked with this decorator, an exception will be raised during initialization.

This is useful for development and testing, where you may want to run only a specific app without modifying configuration files.

Source code in src/hassette/app/app.py

def only_app(app_cls: type[AppT]) -> type[AppT]:
    """Decorator to mark an app class as the only one to run. If more than one app is marked with this decorator,
    an exception will be raised during initialization.

    This is useful for development and testing, where you may want to run only a specific app without
    modifying configuration files.
    """
    app_cls._only_app = True
    return app_cls