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.

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."""

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

    source_tier: ClassVar[SourceTier] = "app"

    app_manifest: ClassVar[AppManifest]
    "Manifest for the app itself, not used by app instances."

    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: logging.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."""

    def __init__(
        self,
        hassette: "Hassette",
        *,
        app_config: AppConfigT,
        index: int,
        api_factory: type[Resource] | None = None,
        parent: Resource | 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
        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)

    @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."""
        return self.app_manifest.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

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

    @final
    async def cleanup(self, timeout: int | 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)

role: ResourceRole = ResourceRole.APP class-attribute

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

app_manifest: AppManifest class-attribute

Manifest for the app itself, not used by app instances.

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: logging.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.

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.

instance_name: str property

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

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()

cleanup(timeout: int | 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: int | 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)

AppSync

Bases: App[AppConfigT]

Synchronous adapter for App.

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

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

    @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 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 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
    )

    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."""

    @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.

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