Index

AppManifest

Bases: ExcludeExtrasMixin, BaseModel

Manifest for a Hassette app.

Source code in src/hassette/config/classes.py

class AppManifest(ExcludeExtrasMixin, BaseModel):
    """Manifest for a Hassette app."""

    model_config = ConfigDict(
        extra="allow", coerce_numbers_to_str=True, validate_assignment=True, use_attribute_docstrings=True
    )

    app_key: str = Field(default=...)
    """Reflects the key for this app in hassette.toml"""

    enabled: bool = Field(default=True)
    """Whether the app is enabled or not, will default to True if not set. Does not consider @only_app decorator."""

    filename: str = Field(default=..., examples=["my_app.py"], validation_alias=AliasChoices("filename", "file_name"))
    """Filename of the app, will be looked for in app_path"""

    class_name: str = Field(
        default=..., examples=["MyApp"], validation_alias=AliasChoices("class_name", "class", "module", "module_name")
    )
    """Class name of the app"""

    display_name: str = Field(default=..., examples=["My App"])
    """Display name of the app, will use class_name if not set"""

    app_dir: Path = Field(..., examples=["./apps"])
    """Path to the app directory, relative to current working directory or absolute"""

    app_config: dict[str, Any] | list[dict[str, Any]] = Field(
        default_factory=dict, validation_alias=AliasChoices("config", "app_config"), validate_default=True
    )
    """Instance configuration for the app"""

    auto_loaded: bool = Field(default=False)
    """Whether the app was auto-detected or manually configured"""

    full_path: Path
    """Fully resolved path to the app file"""

    def __repr__(self) -> str:
        return f"<AppManifest {self.display_name} ({self.class_name}) - enabled={self.enabled} file={self.filename}>"

    @model_validator(mode="before")
    @classmethod
    def validate_app_manifest(cls, values: dict[str, Any]) -> dict[str, Any]:
        """Validate the app configuration."""
        required_keys = ["filename", "class_name", "app_dir"]
        missing_keys = [key for key in required_keys if key not in values]
        if missing_keys:
            raise ValueError(f"App configuration is missing required keys: {', '.join(missing_keys)}")

        values["app_dir"] = app_dir = Path(values["app_dir"]).resolve()

        values["display_name"] = values.get("display_name") or values.get("class_name")

        if app_dir.is_file():
            LOGGER.warning("App directory %s is a file, using the parent directory as app_dir", app_dir)
            values["filename"] = app_dir.name
            values["app_dir"] = app_dir.parent

        return values

    @field_validator("app_config", mode="before")
    @classmethod
    def validate_app_config(cls, v: Any, validation_info: ValidationInfo) -> Any:
        """Set instance name if not set in config."""

        if not v:
            return v

        if isinstance(v, dict):
            v = [v]

        class_name = validation_info.data.get("class_name", "UnknownApp")

        for idx, item in enumerate(v):
            if "instance_name" not in item or not item["instance_name"]:
                item["instance_name"] = f"{class_name}.{idx}"

        return v

    def validate_model_extra(self) -> None:
        if not self.model_extra:
            return

        keys = list(self.model_extra.keys())
        msg = (
            f"{type(self).__name__} - {self.display_name} - Instance configuration values should be"
            " set under the `config` field:\n"
            f"  {keys}\n"
            "This will ensure proper validation and handling of custom configurations."
        )

        if not self.app_config:
            self.app_config = deepcopy(self.model_extra)
        elif isinstance(self.app_config, dict) and not set(self.app_config).intersection(set(keys)):
            self.app_config.update(deepcopy(self.model_extra))
        else:
            if isinstance(self.app_config, list):
                msg += "\nNote: Unable to merge extra fields into list-based config."
            elif isinstance(self.app_config, dict):
                msg += "\nNote: Unable to merge extra fields into existing config due to intersecting keys."

            msg += "\nExtra fields will be ignored. Please update your configuration."

        warn(msg, stacklevel=5)

    def model_post_init(self, context: Any) -> None:
        self.validate_model_extra()

        # if we don't have app_config then we don't have any apps with config
        # which means we have, at most, one app
        # so we can just set the default instance name
        if not self.app_config:
            self.app_config = [{"instance_name": f"{self.class_name}.0"}]

app_key: str = Field(default=...) class-attribute instance-attribute

Reflects the key for this app in hassette.toml

enabled: bool = Field(default=True) class-attribute instance-attribute

Whether the app is enabled or not, will default to True if not set. Does not consider @only_app decorator.

filename: str = Field(default=..., examples=['my_app.py'], validation_alias=(AliasChoices('filename', 'file_name'))) class-attribute instance-attribute

Filename of the app, will be looked for in app_path

class_name: str = Field(default=..., examples=['MyApp'], validation_alias=(AliasChoices('class_name', 'class', 'module', 'module_name'))) class-attribute instance-attribute

Class name of the app

display_name: str = Field(default=..., examples=['My App']) class-attribute instance-attribute

Display name of the app, will use class_name if not set

app_dir: Path = Field(..., examples=['./apps']) class-attribute instance-attribute

Path to the app directory, relative to current working directory or absolute

app_config: dict[str, Any] | list[dict[str, Any]] = Field(default_factory=dict, validation_alias=(AliasChoices('config', 'app_config')), validate_default=True) class-attribute instance-attribute

Instance configuration for the app

auto_loaded: bool = Field(default=False) class-attribute instance-attribute

Whether the app was auto-detected or manually configured

full_path: Path instance-attribute

Fully resolved path to the app file

validate_app_manifest(values: dict[str, Any]) -> dict[str, Any] classmethod

Validate the app configuration.

Source code in src/hassette/config/classes.py

@model_validator(mode="before")
@classmethod
def validate_app_manifest(cls, values: dict[str, Any]) -> dict[str, Any]:
    """Validate the app configuration."""
    required_keys = ["filename", "class_name", "app_dir"]
    missing_keys = [key for key in required_keys if key not in values]
    if missing_keys:
        raise ValueError(f"App configuration is missing required keys: {', '.join(missing_keys)}")

    values["app_dir"] = app_dir = Path(values["app_dir"]).resolve()

    values["display_name"] = values.get("display_name") or values.get("class_name")

    if app_dir.is_file():
        LOGGER.warning("App directory %s is a file, using the parent directory as app_dir", app_dir)
        values["filename"] = app_dir.name
        values["app_dir"] = app_dir.parent

    return values

validate_app_config(v: Any, validation_info: ValidationInfo) -> Any classmethod

Set instance name if not set in config.

Source code in src/hassette/config/classes.py

@field_validator("app_config", mode="before")
@classmethod
def validate_app_config(cls, v: Any, validation_info: ValidationInfo) -> Any:
    """Set instance name if not set in config."""

    if not v:
        return v

    if isinstance(v, dict):
        v = [v]

    class_name = validation_info.data.get("class_name", "UnknownApp")

    for idx, item in enumerate(v):
        if "instance_name" not in item or not item["instance_name"]:
            item["instance_name"] = f"{class_name}.{idx}"

    return v

HassetteConfig

Bases: ExcludeExtrasMixin, BaseSettings

Configuration for Hassette.

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

class HassetteConfig(ExcludeExtrasMixin, BaseSettings):
    """Configuration for Hassette."""

    model_config = SettingsConfigDict(
        env_prefix="hassette__",
        env_file=ENV_FILE_LOCATIONS,
        toml_file=TOML_FILE_LOCATIONS,
        env_ignore_empty=True,
        extra="allow",
        env_nested_delimiter="__",
        coerce_numbers_to_str=True,
        validate_by_name=True,
        use_attribute_docstrings=True,
        validate_assignment=True,
        cli_parse_args=False,
        nested_model_default_partial_update=True,
    )

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: type["BaseSettings"],
        init_settings: PydanticBaseSettingsSource,
        env_settings: PydanticBaseSettingsSource,
        dotenv_settings: PydanticBaseSettingsSource,
        file_secret_settings: PydanticBaseSettingsSource,
    ) -> tuple[PydanticBaseSettingsSource, ...]:
        sources = (
            init_settings,
            env_settings,
            dotenv_settings,
            file_secret_settings,
            HassetteTomlConfigSettingsSource(settings_cls),
        )
        return sources

    database: DatabaseConfig = Field(default_factory=DatabaseConfig)
    """Database storage, retention, and operational settings."""

    websocket: WebSocketConfig = Field(default_factory=WebSocketConfig)
    """WebSocket connection, retry, and recovery timing settings."""

    logging: LoggingConfig = Field(default_factory=LoggingConfig)
    """Logging level, format, queue, and per-service log-level settings."""

    lifecycle: LifecycleConfig = Field(default_factory=LifecycleConfig)
    """Startup, shutdown, and per-operation timeout settings."""

    web_api: WebApiConfig = Field(default_factory=WebApiConfig)
    """Web API and UI server settings."""

    apps: AppsConfig = Field(default_factory=AppsConfig)
    """App directory, auto-detection, and manifest settings."""

    scheduler: SchedulerConfig = Field(default_factory=SchedulerConfig)
    """Scheduler delay, threshold, and job-timeout settings."""

    file_watcher: FileWatcherConfig = Field(default_factory=FileWatcherConfig)
    """File watcher debounce, step, and enable/disable settings."""

    # note - not actually used here, reflects the --config-file / --env-file flags on the cyclopts default command
    config_file: Path | str | None = Field(default=Path("hassette.toml"))
    """Path to the configuration file."""

    # note - not actually used here, reflects the --config-file / --env-file flags on the cyclopts default command
    env_file: Path | str | None = Field(default=Path(".env"))
    """Path to the environment file."""

    dev_mode: bool = Field(default_factory=get_dev_mode)
    """Enable developer mode, which may include additional logging and features."""

    # Home Assistant connection — cross-cutting
    base_url: str = Field(default="http://127.0.0.1:8123")
    """Base URL of the Home Assistant instance"""

    verify_ssl: bool = Field(default=True)
    """Whether to verify SSL certificates when connecting to Home Assistant. Useful to disable for self-signed
    certificates."""

    token: str | None = Field(
        default=None,
        validation_alias=AliasChoices("token", "hassette__token", "ha_token", "home_assistant_token"),
    )
    """Access token for Home Assistant instance"""

    config_dir: Path = Field(default_factory=default_config_dir)
    """Directory to load/save configuration."""

    data_dir: Path = Field(default_factory=default_data_dir)
    """Directory to store Hassette data."""

    import_dot_env_files: bool = Field(default=True)
    """Whether to import .env files specified in env_files. With this disabled, the .env file provided will only
    be used for loading settings. With this enabled, the .env files will also be loaded into os.environ."""

    run_app_precheck: bool = Field(default=True)
    """Whether to run the app precheck before starting Hassette. This is recommended, but if any apps fail to load
    then Hassette will not start."""

    allow_startup_if_app_precheck_fails: bool = Field(default=False)
    """Whether to allow Hassette to start even if the app precheck fails. This is generally not recommended."""

    hassette_event_buffer_size: int = Field(default=1000)
    """Buffer capacity of the internal anyio memory channel used to route events to the bus."""

    default_cache_size: int = Field(default=DEFAULT_CACHE_SIZE_BYTES)
    """Default size limit for caches in bytes. Defaults to 100 MiB."""

    strict_lifecycle: bool = Field(default=False)
    """Enable strict validation for lifecycle transitions, connection state, and registries.

    Controls three subsystems uniformly:
    - Resource lifecycle: invalid ResourceStatus transitions raise InvalidLifecycleTransitionError
    - WebSocket connection: invalid ConnectionState transitions raise InvalidLifecycleTransitionError
    - Registry validation: startup issues raise RegistryValidationError

    When False (default), all three subsystems log WARNING instead of raising.
    The test harness sets this to True by default."""

    asyncio_debug_mode: bool = Field(default=False)
    """Whether to enable asyncio debug mode."""

    state_proxy_poll_interval_seconds: int = Field(default=30)
    """Interval in seconds to poll the state proxy for updates."""

    disable_state_proxy_polling: bool = Field(default=False)
    """Whether to disable polling for the state proxy. Defaults to False."""

    bus_excluded_domains: tuple[str, ...] = Field(default_factory=tuple)
    """Domains whose events should be skipped by the bus; supports glob patterns (e.g. 'sensor', 'media_*')."""

    bus_excluded_entities: tuple[str, ...] = Field(default_factory=tuple)
    """Entity IDs whose events should be skipped by the bus; supports glob patterns."""

    allow_reload_in_prod: bool = Field(default=False)
    """Whether to enable the file watcher for automatic app reloads in production mode.

    When True, file changes trigger automatic app reloads (same as dev_mode).
    Manual app management (start/stop/reload via API) is always available
    regardless of this setting. Defaults to False.
    """

    allow_only_app_in_prod: bool = Field(default=False)
    """Whether to allow the `only_app` decorator in production mode. Defaults to False."""

    @property
    def env_files(self) -> set[Path]:
        """Return a list of environment files that Pydantic will check."""
        return filter_paths_to_unique_existing(self.model_config.get("env_file", []))

    @property
    def toml_files(self) -> set[Path]:
        """Return a list of toml files that Pydantic will check."""
        return filter_paths_to_unique_existing(self.model_config.get("toml_file", []))

    def get_watchable_files(self) -> set[Path]:
        """Return a list of files to watch for changes."""

        files = self.env_files | self.toml_files
        files.add(self.apps.directory.resolve())

        # just add everything from here, since we'll filter it to only existing and remove duplicates later
        for app_manifest in self.apps.manifests.values():
            with suppress(FileNotFoundError):
                files.add(app_manifest.full_path)
                files.add(app_manifest.app_dir)

        files = filter_paths_to_unique_existing(files)

        return files

    @property
    def auth_headers(self) -> dict[str, str]:
        """Return the headers required for authentication."""
        if self.token is None:
            return {}
        return {"Authorization": f"Bearer {self.token}"}

    @property
    def headers(self) -> dict[str, str]:
        """Return the headers for API requests."""
        return {**self.auth_headers, "Content-Type": "application/json"}

    @property
    def truncated_token(self) -> str:
        """Return a truncated version of the token for display purposes."""
        if self.token is None:
            return "<not set>"
        if len(self.token) < TOKEN_SHORT_THRESHOLD:
            return "***"
        if len(self.token) <= TOKEN_MEDIUM_THRESHOLD:
            return f"{self.token[:TOKEN_SHORT_PREFIX_LENGTH]}***"
        return f"{self.token[:TOKEN_LONG_PREFIX_LENGTH]}...{self.token[-TOKEN_LONG_PREFIX_LENGTH:]}"

    @model_validator(mode="after")
    def validate_log_retention_days(self) -> "HassetteConfig":
        """Ensure logging.log_retention_days <= database.retention_days.

        Log records reference executions; allowing log records to outlive
        the execution records that produced them would break referential
        integrity semantics even though the FK is not enforced at the DB level.
        """
        if self.logging.log_retention_days > self.database.retention_days:
            raise ValueError(
                f"logging.log_retention_days ({self.logging.log_retention_days}) must be <= "
                f"database.retention_days ({self.database.retention_days})"
            )
        return self

    @field_validator("config_dir", "data_dir", mode="after")
    @classmethod
    def resolve_paths(cls, value: Path) -> Path:
        """Resolve paths to absolute without creating directories.

        Directory creation is deferred to server startup (Hassette.wire_services)
        so that read-only CLI commands don't produce filesystem side effects.
        """
        return value.resolve()

    def ensure_directories(self) -> None:
        """Create config_dir and data_dir if they don't exist."""
        for directory in (self.config_dir, self.data_dir):
            if not directory.exists():
                LOGGER.debug("Creating directory %s as it does not exist", directory)
                directory.mkdir(parents=True, exist_ok=True)

    def reload(self) -> None:
        """Reload the configuration from all sources."""
        # see: https://docs.pydantic.dev/latest/concepts/pydantic_settings/#in-place-reloading
        self.__init__()
        self.set_validated_app_manifests()

    def model_post_init(self, *args: Any) -> None:
        """Set default values for any unset fields after initialization."""
        # Snapshot which group fields were set by actual user sources (init kwargs, env vars, TOML)
        # BEFORE the defaults loop runs — setattr in the defaults loop updates model_fields_set,
        # which would incorrectly block legacy migrations for defaulted fields.
        pre_migration_fields: dict[str, set[str]] = {
            name: set(getattr(self, name).model_fields_set) for name in NESTED_GROUPS
        }

        default_str = "default (dev)" if self.dev_mode else "default (prod)"
        defaults = get_defaults_dict(dev=self.dev_mode)

        # Apply root-level flat defaults (e.g. dev_mode, allow_startup_if_app_precheck_fails,
        # state_proxy_poll_interval_seconds)
        for fname in type(self).model_fields:
            if fname in self.model_fields_set or fname not in defaults:
                continue
            # Skip nested group names — they are handled below
            if fname in NESTED_GROUPS:
                continue
            default_value = defaults[fname]
            LOGGER.debug("Setting %s for unset field %s: %s", default_str, fname, default_value)
            setattr(self, fname, default_value)

        # Apply nested group defaults (e.g. [hassette.websocket], [hassette.scheduler])
        for group_name in NESTED_GROUPS:
            if group_name not in defaults or group_name in self.model_fields_set:
                continue
            group_defaults = defaults[group_name]
            if not isinstance(group_defaults, dict):
                continue
            group_obj = getattr(self, group_name)
            for sub_field, sub_value in group_defaults.items():
                LOGGER.debug(
                    "Setting %s for unset nested field %s.%s: %s",
                    default_str,
                    group_name,
                    sub_field,
                    sub_value,
                )
                setattr(group_obj, sub_field, sub_value)

        legacy_extra_values: dict[str, object] = {}
        if self.model_extra:
            if "app" in self.model_extra:
                LOGGER.warning(
                    "Detected legacy [hassette.app] section — this key is ignored. "
                    "Rename to [hassette.apps] and move app definitions from "
                    "[hassette.app.apps.<name>] to [hassette.apps.<name>]. "
                    "Environment variables: HASSETTE__APP__* -> HASSETTE__APPS__*."
                )

            legacy_hits = {k: LEGACY_KEY_MIGRATION[k] for k in self.model_extra if k in LEGACY_KEY_MIGRATION}
            if legacy_hits:
                legacy_extra_values = {k: self.model_extra[k] for k in legacy_hits}
                self.apply_legacy_migrations(legacy_hits, pre_migration_fields)

        self.apply_legacy_env_vars(pre_migration_fields, legacy_extra_values)

    def apply_legacy_migrations(self, legacy_hits: dict[str, str], pre_migration_fields: dict[str, set[str]]) -> None:
        migrations: dict[str, dict[str, object]] = {}
        for old_key, dot_path in legacy_hits.items():
            group_name, sub_field = dot_path.split(".", 1)
            if sub_field in pre_migration_fields.get(group_name, set()):
                continue
            LOGGER.warning(
                "Migrating legacy config key %r → %s (source: config file). "
                "Update your configuration to use HASSETTE__%s instead.",
                old_key,
                dot_path,
                dot_path.replace(".", "__").upper(),
            )
            migrations.setdefault(group_name, {})[sub_field] = self.model_extra[old_key]  # pyright: ignore[reportOptionalSubscript]
        self.apply_group_updates(migrations)

    def apply_legacy_env_vars(
        self, pre_migration_fields: dict[str, set[str]], legacy_extra_values: dict[str, object]
    ) -> None:
        env_prefix = self.model_config.get("env_prefix", "").upper()
        migrations: dict[str, dict[str, object]] = {}
        for old_key, dot_path in LEGACY_KEY_MIGRATION.items():
            env_var = f"{env_prefix}{old_key.upper()}"
            raw_value = os.environ.get(env_var)
            if not raw_value:  # None = not set, "" = empty (matches env_ignore_empty)
                continue
            # Skip if apply_legacy_migrations already handled this key with the same value
            # (both came from the same env var via pydantic-settings model_extra). When the
            # values differ, env should override TOML — that's the priority fix.
            if old_key in legacy_extra_values and str(legacy_extra_values[old_key]) == raw_value:
                continue
            group_name, sub_field = dot_path.split(".", 1)
            if sub_field in pre_migration_fields.get(group_name, set()):
                continue
            LOGGER.warning(
                "Migrating legacy env var %s → %s. Update your configuration to use HASSETTE__%s instead.",
                env_var,
                dot_path,
                dot_path.replace(".", "__").upper(),
            )
            migrations.setdefault(group_name, {})[sub_field] = raw_value
        self.apply_group_updates(migrations)

    def apply_group_updates(self, migrations: dict[str, dict[str, object]]) -> None:
        for group_name, updates in migrations.items():
            group_obj = getattr(self, group_name)
            group_data = group_obj.model_dump()
            group_data.update(updates)
            setattr(self, group_name, type(group_obj).model_validate(group_data))

    @classmethod
    def get_config(cls) -> "HassetteConfig":
        """Get the global configuration instance.

        Raises:
            HassetteNotInitializedError: If the Hassette instance is not initialized.
        """
        return ctx.get_hassette_config()

    def set_validated_app_manifests(self):
        """Cleans up and validates the apps configuration, including auto-detection."""
        cleaned_apps_dict: dict[str, AppDict] = {}

        # track known paths to simplify dupe detection during auto-detect
        known_paths: set[Path] = set()

        for k, v in self.apps.apps.copy().items():
            if not isinstance(v, dict):
                continue
            try:
                v = clean_app(k, v, self.apps.directory)
            except (KeyError, TypeError):
                LOGGER.warning("Skipping app %r: missing required keys (filename or class_name)", k)
                continue
            cleaned_apps_dict[k] = v

            # track known paths
            known_paths.add(v["full_path"])

        if self.apps.autodetect:
            autodetected_apps = autodetect_apps(self.apps.directory, known_paths, set(self.apps.exclude_dirs))
            for k, v in autodetected_apps.items():
                app_dir = v["app_dir"]
                full_path = app_dir / v["filename"]
                LOGGER.debug("Auto-detected app %s from %s", k, full_path)
                if k in cleaned_apps_dict:
                    LOGGER.debug("Skipping auto-detected app %s as it conflicts with manually configured app", k)
                    continue
                cleaned_apps_dict[k] = v
                known_paths.add(full_path.resolve())

        app_manifest_dict: dict[str, AppManifest] = {}
        for k, v in cleaned_apps_dict.items():
            if is_framework_key(k):
                raise ValueError(
                    f"App key {k!r} is reserved for framework internals "
                    f"(reserved prefix: '{FRAMEWORK_APP_KEY_PREFIX}'). "
                    f"Rename the app in your configuration (source: {v.get('full_path', 'unknown')})."
                )
            app_manifest_dict[k] = AppManifest.model_validate(v)

        self.apps.manifests = app_manifest_dict

database: DatabaseConfig = Field(default_factory=DatabaseConfig) class-attribute instance-attribute

Database storage, retention, and operational settings.

websocket: WebSocketConfig = Field(default_factory=WebSocketConfig) class-attribute instance-attribute

WebSocket connection, retry, and recovery timing settings.

logging: LoggingConfig = Field(default_factory=LoggingConfig) class-attribute instance-attribute

Logging level, format, queue, and per-service log-level settings.

lifecycle: LifecycleConfig = Field(default_factory=LifecycleConfig) class-attribute instance-attribute

Startup, shutdown, and per-operation timeout settings.

web_api: WebApiConfig = Field(default_factory=WebApiConfig) class-attribute instance-attribute

Web API and UI server settings.

apps: AppsConfig = Field(default_factory=AppsConfig) class-attribute instance-attribute

App directory, auto-detection, and manifest settings.

scheduler: SchedulerConfig = Field(default_factory=SchedulerConfig) class-attribute instance-attribute

Scheduler delay, threshold, and job-timeout settings.

file_watcher: FileWatcherConfig = Field(default_factory=FileWatcherConfig) class-attribute instance-attribute

File watcher debounce, step, and enable/disable settings.

config_file: Path | str | None = Field(default=(Path('hassette.toml'))) class-attribute instance-attribute

Path to the configuration file.

env_file: Path | str | None = Field(default=(Path('.env'))) class-attribute instance-attribute

Path to the environment file.

dev_mode: bool = Field(default_factory=get_dev_mode) class-attribute instance-attribute

Enable developer mode, which may include additional logging and features.

base_url: str = Field(default='http://127.0.0.1:8123') class-attribute instance-attribute

Base URL of the Home Assistant instance

verify_ssl: bool = Field(default=True) class-attribute instance-attribute

Whether to verify SSL certificates when connecting to Home Assistant. Useful to disable for self-signed certificates.

token: str | None = Field(default=None, validation_alias=(AliasChoices('token', 'hassette__token', 'ha_token', 'home_assistant_token'))) class-attribute instance-attribute

Access token for Home Assistant instance

config_dir: Path = Field(default_factory=default_config_dir) class-attribute instance-attribute

Directory to load/save configuration.

data_dir: Path = Field(default_factory=default_data_dir) class-attribute instance-attribute

Directory to store Hassette data.

import_dot_env_files: bool = Field(default=True) class-attribute instance-attribute

Whether to import .env files specified in env_files. With this disabled, the .env file provided will only be used for loading settings. With this enabled, the .env files will also be loaded into os.environ.

run_app_precheck: bool = Field(default=True) class-attribute instance-attribute

Whether to run the app precheck before starting Hassette. This is recommended, but if any apps fail to load then Hassette will not start.

allow_startup_if_app_precheck_fails: bool = Field(default=False) class-attribute instance-attribute

Whether to allow Hassette to start even if the app precheck fails. This is generally not recommended.

hassette_event_buffer_size: int = Field(default=1000) class-attribute instance-attribute

Buffer capacity of the internal anyio memory channel used to route events to the bus.

default_cache_size: int = Field(default=DEFAULT_CACHE_SIZE_BYTES) class-attribute instance-attribute

Default size limit for caches in bytes. Defaults to 100 MiB.

strict_lifecycle: bool = Field(default=False) class-attribute instance-attribute

Enable strict validation for lifecycle transitions, connection state, and registries.

Controls three subsystems uniformly: - Resource lifecycle: invalid ResourceStatus transitions raise InvalidLifecycleTransitionError - WebSocket connection: invalid ConnectionState transitions raise InvalidLifecycleTransitionError - Registry validation: startup issues raise RegistryValidationError

When False (default), all three subsystems log WARNING instead of raising. The test harness sets this to True by default.

asyncio_debug_mode: bool = Field(default=False) class-attribute instance-attribute

Whether to enable asyncio debug mode.

state_proxy_poll_interval_seconds: int = Field(default=30) class-attribute instance-attribute

Interval in seconds to poll the state proxy for updates.

disable_state_proxy_polling: bool = Field(default=False) class-attribute instance-attribute

Whether to disable polling for the state proxy. Defaults to False.

bus_excluded_domains: tuple[str, ...] = Field(default_factory=tuple) class-attribute instance-attribute

Domains whose events should be skipped by the bus; supports glob patterns (e.g. 'sensor', 'media_*').

bus_excluded_entities: tuple[str, ...] = Field(default_factory=tuple) class-attribute instance-attribute

Entity IDs whose events should be skipped by the bus; supports glob patterns.

allow_reload_in_prod: bool = Field(default=False) class-attribute instance-attribute

Whether to enable the file watcher for automatic app reloads in production mode.

When True, file changes trigger automatic app reloads (same as dev_mode). Manual app management (start/stop/reload via API) is always available regardless of this setting. Defaults to False.

allow_only_app_in_prod: bool = Field(default=False) class-attribute instance-attribute

Whether to allow the only_app decorator in production mode. Defaults to False.

env_files: set[Path] property

Return a list of environment files that Pydantic will check.

toml_files: set[Path] property

Return a list of toml files that Pydantic will check.

auth_headers: dict[str, str] property

Return the headers required for authentication.

headers: dict[str, str] property

Return the headers for API requests.

truncated_token: str property

Return a truncated version of the token for display purposes.

get_watchable_files() -> set[Path]

Return a list of files to watch for changes.

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

def get_watchable_files(self) -> set[Path]:
    """Return a list of files to watch for changes."""

    files = self.env_files | self.toml_files
    files.add(self.apps.directory.resolve())

    # just add everything from here, since we'll filter it to only existing and remove duplicates later
    for app_manifest in self.apps.manifests.values():
        with suppress(FileNotFoundError):
            files.add(app_manifest.full_path)
            files.add(app_manifest.app_dir)

    files = filter_paths_to_unique_existing(files)

    return files

validate_log_retention_days() -> HassetteConfig

Ensure logging.log_retention_days <= database.retention_days.

Log records reference executions; allowing log records to outlive the execution records that produced them would break referential integrity semantics even though the FK is not enforced at the DB level.

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

@model_validator(mode="after")
def validate_log_retention_days(self) -> "HassetteConfig":
    """Ensure logging.log_retention_days <= database.retention_days.

    Log records reference executions; allowing log records to outlive
    the execution records that produced them would break referential
    integrity semantics even though the FK is not enforced at the DB level.
    """
    if self.logging.log_retention_days > self.database.retention_days:
        raise ValueError(
            f"logging.log_retention_days ({self.logging.log_retention_days}) must be <= "
            f"database.retention_days ({self.database.retention_days})"
        )
    return self

resolve_paths(value: Path) -> Path classmethod

Resolve paths to absolute without creating directories.

Directory creation is deferred to server startup (Hassette.wire_services) so that read-only CLI commands don't produce filesystem side effects.

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

@field_validator("config_dir", "data_dir", mode="after")
@classmethod
def resolve_paths(cls, value: Path) -> Path:
    """Resolve paths to absolute without creating directories.

    Directory creation is deferred to server startup (Hassette.wire_services)
    so that read-only CLI commands don't produce filesystem side effects.
    """
    return value.resolve()

ensure_directories() -> None

Create config_dir and data_dir if they don't exist.

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

def ensure_directories(self) -> None:
    """Create config_dir and data_dir if they don't exist."""
    for directory in (self.config_dir, self.data_dir):
        if not directory.exists():
            LOGGER.debug("Creating directory %s as it does not exist", directory)
            directory.mkdir(parents=True, exist_ok=True)

reload() -> None

Reload the configuration from all sources.

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

def reload(self) -> None:
    """Reload the configuration from all sources."""
    # see: https://docs.pydantic.dev/latest/concepts/pydantic_settings/#in-place-reloading
    self.__init__()
    self.set_validated_app_manifests()

model_post_init(*args: Any) -> None

Set default values for any unset fields after initialization.

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

def model_post_init(self, *args: Any) -> None:
    """Set default values for any unset fields after initialization."""
    # Snapshot which group fields were set by actual user sources (init kwargs, env vars, TOML)
    # BEFORE the defaults loop runs — setattr in the defaults loop updates model_fields_set,
    # which would incorrectly block legacy migrations for defaulted fields.
    pre_migration_fields: dict[str, set[str]] = {
        name: set(getattr(self, name).model_fields_set) for name in NESTED_GROUPS
    }

    default_str = "default (dev)" if self.dev_mode else "default (prod)"
    defaults = get_defaults_dict(dev=self.dev_mode)

    # Apply root-level flat defaults (e.g. dev_mode, allow_startup_if_app_precheck_fails,
    # state_proxy_poll_interval_seconds)
    for fname in type(self).model_fields:
        if fname in self.model_fields_set or fname not in defaults:
            continue
        # Skip nested group names — they are handled below
        if fname in NESTED_GROUPS:
            continue
        default_value = defaults[fname]
        LOGGER.debug("Setting %s for unset field %s: %s", default_str, fname, default_value)
        setattr(self, fname, default_value)

    # Apply nested group defaults (e.g. [hassette.websocket], [hassette.scheduler])
    for group_name in NESTED_GROUPS:
        if group_name not in defaults or group_name in self.model_fields_set:
            continue
        group_defaults = defaults[group_name]
        if not isinstance(group_defaults, dict):
            continue
        group_obj = getattr(self, group_name)
        for sub_field, sub_value in group_defaults.items():
            LOGGER.debug(
                "Setting %s for unset nested field %s.%s: %s",
                default_str,
                group_name,
                sub_field,
                sub_value,
            )
            setattr(group_obj, sub_field, sub_value)

    legacy_extra_values: dict[str, object] = {}
    if self.model_extra:
        if "app" in self.model_extra:
            LOGGER.warning(
                "Detected legacy [hassette.app] section — this key is ignored. "
                "Rename to [hassette.apps] and move app definitions from "
                "[hassette.app.apps.<name>] to [hassette.apps.<name>]. "
                "Environment variables: HASSETTE__APP__* -> HASSETTE__APPS__*."
            )

        legacy_hits = {k: LEGACY_KEY_MIGRATION[k] for k in self.model_extra if k in LEGACY_KEY_MIGRATION}
        if legacy_hits:
            legacy_extra_values = {k: self.model_extra[k] for k in legacy_hits}
            self.apply_legacy_migrations(legacy_hits, pre_migration_fields)

    self.apply_legacy_env_vars(pre_migration_fields, legacy_extra_values)

get_config() -> HassetteConfig classmethod

Get the global configuration instance.

Raises:

Type	Description
HassetteNotInitializedError	If the Hassette instance is not initialized.

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

@classmethod
def get_config(cls) -> "HassetteConfig":
    """Get the global configuration instance.

    Raises:
        HassetteNotInitializedError: If the Hassette instance is not initialized.
    """
    return ctx.get_hassette_config()

set_validated_app_manifests()

Cleans up and validates the apps configuration, including auto-detection.

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

def set_validated_app_manifests(self):
    """Cleans up and validates the apps configuration, including auto-detection."""
    cleaned_apps_dict: dict[str, AppDict] = {}

    # track known paths to simplify dupe detection during auto-detect
    known_paths: set[Path] = set()

    for k, v in self.apps.apps.copy().items():
        if not isinstance(v, dict):
            continue
        try:
            v = clean_app(k, v, self.apps.directory)
        except (KeyError, TypeError):
            LOGGER.warning("Skipping app %r: missing required keys (filename or class_name)", k)
            continue
        cleaned_apps_dict[k] = v

        # track known paths
        known_paths.add(v["full_path"])

    if self.apps.autodetect:
        autodetected_apps = autodetect_apps(self.apps.directory, known_paths, set(self.apps.exclude_dirs))
        for k, v in autodetected_apps.items():
            app_dir = v["app_dir"]
            full_path = app_dir / v["filename"]
            LOGGER.debug("Auto-detected app %s from %s", k, full_path)
            if k in cleaned_apps_dict:
                LOGGER.debug("Skipping auto-detected app %s as it conflicts with manually configured app", k)
                continue
            cleaned_apps_dict[k] = v
            known_paths.add(full_path.resolve())

    app_manifest_dict: dict[str, AppManifest] = {}
    for k, v in cleaned_apps_dict.items():
        if is_framework_key(k):
            raise ValueError(
                f"App key {k!r} is reserved for framework internals "
                f"(reserved prefix: '{FRAMEWORK_APP_KEY_PREFIX}'). "
                f"Rename the app in your configuration (source: {v.get('full_path', 'unknown')})."
            )
        app_manifest_dict[k] = AppManifest.model_validate(v)

    self.apps.manifests = app_manifest_dict

AppsConfig

Bases: ExcludeExtrasMixin, BaseModel

App directory, auto-detection, exclusion, manifest, and raw-app-dict settings.

In TOML, app definitions live alongside these settings under [hassette.apps]:

.. code-block:: toml

[hassette.apps]
directory = "apps"

[hassette.apps.my_app]
filename = "my_app.py"
class_name = "MyApp"

The model_validator separates known config fields from app-definition dicts so both coexist in the same TOML section.

Source code in src/hassette/config/models.py

class AppsConfig(ExcludeExtrasMixin, BaseModel):
    """App directory, auto-detection, exclusion, manifest, and raw-app-dict settings.

    In TOML, app definitions live alongside these settings under ``[hassette.apps]``:

    .. code-block:: toml

        [hassette.apps]
        directory = "apps"

        [hassette.apps.my_app]
        filename = "my_app.py"
        class_name = "MyApp"

    The ``model_validator`` separates known config fields from app-definition
    dicts so both coexist in the same TOML section.
    """

    autodetect: bool = Field(default=True)
    """Whether to automatically detect apps in the app directory."""

    extend_exclude_dirs: tuple[str, ...] = Field(default_factory=tuple)
    """Additional directories to exclude when auto-detecting apps in the app directory."""

    exclude_dirs: tuple[str, ...] = Field(
        default_factory=lambda data: (
            *data.get("extend_exclude_dirs", ()),
            *AUTODETECT_EXCLUDE_DIRS_DEFAULT,
        )
    )
    """Directories to exclude when auto-detecting apps. Prefer extend_exclude_dirs to avoid
    removing the defaults."""

    manifests: dict[str, AppManifest] = Field(default_factory=dict)
    """Validated app manifests, keyed by app name."""

    apps: dict[str, RawAppDict] = Field(default_factory=dict)
    """Raw configuration for Hassette apps, keyed by app name."""

    directory: Path = Field(default_factory=lambda: Path.cwd() / "apps")
    """Directory to load user apps from."""

    @model_validator(mode="before")
    @classmethod
    def extract_app_definitions(cls, data: Any) -> Any:
        """Separate app-definition dicts from known config fields."""
        if not isinstance(data, dict):
            return data
        data = dict(data)
        known = set(cls.model_fields)
        reserved = known - {"apps", "manifests"}
        app_defs: dict[str, Any] = {}
        for key in list(data):
            if key in known:
                if isinstance(data[key], dict) and key in reserved:
                    raise ValueError(
                        f"App name {key!r} conflicts with a reserved config field. "
                        f"Reserved names: {sorted(reserved)}. "
                        f"Rename the app in your configuration."
                    )
                continue
            if isinstance(data[key], dict):
                app_defs[key] = data.pop(key)
        if app_defs:
            existing = data.get("apps")
            if isinstance(existing, dict):
                data["apps"] = {**existing, **app_defs}
            else:
                data["apps"] = app_defs
        return data

    @field_validator("apps", mode="before")
    @classmethod
    def remove_incomplete_apps(cls, value: dict[str, Any]) -> dict[str, Any]:
        """Remove any apps that are missing required fields before validation."""
        missing_required = {k: v for k, v in value.items() if isinstance(v, dict) and not APP_REQUIRED_KEYS.issubset(v)}
        if missing_required:
            LOGGER.warning(
                "The following apps are missing required keys (%s) and will be ignored: %s",
                ", ".join(APP_REQUIRED_KEYS),
                list(missing_required.keys()),
            )
            for k in missing_required:
                value.pop(k)
        return value

autodetect: bool = Field(default=True) class-attribute instance-attribute

Whether to automatically detect apps in the app directory.

extend_exclude_dirs: tuple[str, ...] = Field(default_factory=tuple) class-attribute instance-attribute

Additional directories to exclude when auto-detecting apps in the app directory.

exclude_dirs: tuple[str, ...] = Field(default_factory=(lambda data: (*(data.get('extend_exclude_dirs', ())), *AUTODETECT_EXCLUDE_DIRS_DEFAULT))) class-attribute instance-attribute

Directories to exclude when auto-detecting apps. Prefer extend_exclude_dirs to avoid removing the defaults.

manifests: dict[str, AppManifest] = Field(default_factory=dict) class-attribute instance-attribute

Validated app manifests, keyed by app name.

apps: dict[str, RawAppDict] = Field(default_factory=dict) class-attribute instance-attribute

Raw configuration for Hassette apps, keyed by app name.

directory: Path = Field(default_factory=(lambda: Path.cwd() / 'apps')) class-attribute instance-attribute

Directory to load user apps from.

extract_app_definitions(data: Any) -> Any classmethod

Separate app-definition dicts from known config fields.

Source code in src/hassette/config/models.py

@model_validator(mode="before")
@classmethod
def extract_app_definitions(cls, data: Any) -> Any:
    """Separate app-definition dicts from known config fields."""
    if not isinstance(data, dict):
        return data
    data = dict(data)
    known = set(cls.model_fields)
    reserved = known - {"apps", "manifests"}
    app_defs: dict[str, Any] = {}
    for key in list(data):
        if key in known:
            if isinstance(data[key], dict) and key in reserved:
                raise ValueError(
                    f"App name {key!r} conflicts with a reserved config field. "
                    f"Reserved names: {sorted(reserved)}. "
                    f"Rename the app in your configuration."
                )
            continue
        if isinstance(data[key], dict):
            app_defs[key] = data.pop(key)
    if app_defs:
        existing = data.get("apps")
        if isinstance(existing, dict):
            data["apps"] = {**existing, **app_defs}
        else:
            data["apps"] = app_defs
    return data

remove_incomplete_apps(value: dict[str, Any]) -> dict[str, Any] classmethod

Remove any apps that are missing required fields before validation.

Source code in src/hassette/config/models.py

@field_validator("apps", mode="before")
@classmethod
def remove_incomplete_apps(cls, value: dict[str, Any]) -> dict[str, Any]:
    """Remove any apps that are missing required fields before validation."""
    missing_required = {k: v for k, v in value.items() if isinstance(v, dict) and not APP_REQUIRED_KEYS.issubset(v)}
    if missing_required:
        LOGGER.warning(
            "The following apps are missing required keys (%s) and will be ignored: %s",
            ", ".join(APP_REQUIRED_KEYS),
            list(missing_required.keys()),
        )
        for k in missing_required:
            value.pop(k)
    return value

DatabaseConfig

Bases: ExcludeExtrasMixin, BaseModel

Database storage, retention, write-queue, and operational-interval settings.

Source code in src/hassette/config/models.py

class DatabaseConfig(ExcludeExtrasMixin, BaseModel):
    """Database storage, retention, write-queue, and operational-interval settings."""

    path: Path | None = Field(default=None)
    """Path to the SQLite database file. Defaults to data_dir / "hassette.db" when None."""

    retention_days: int = Field(default=7, ge=1)
    """Number of days to retain execution records in the ``executions`` table."""

    max_size_mb: float = Field(default=500, ge=0)
    """Maximum database file size in MB. When exceeded, oldest execution records are deleted.
    0 disables the size failsafe."""

    migration_timeout_seconds: int = Field(default=120, ge=10)
    """Maximum seconds to wait for SQL schema migrations to complete at startup."""

    write_queue_max: int = Field(default=2000, ge=1)
    """Maximum pending coroutines in the DatabaseService write queue. Bounds memory growth
    under sustained I/O pressure. Fire-and-forget tasks are dropped on overflow; submit()
    callers block until space is available."""

    telemetry_write_queue_max: int = Field(default=1000, ge=1)
    """Maximum pending records in the CommandExecutor write queue before records are dropped."""

    heartbeat_interval_seconds: int = Field(default=300, ge=10)
    """Interval in seconds between database heartbeat checks."""

    retention_interval_seconds: int = Field(default=3600, ge=60)
    """Interval in seconds between retention enforcement runs."""

    size_failsafe_interval_seconds: int = Field(default=3600, ge=60)
    """Interval in seconds between size failsafe enforcement runs."""

    size_failsafe_max_iterations: int = Field(default=10, ge=1)
    """Maximum number of delete-batch iterations per size failsafe run."""

    size_failsafe_delete_batch: int = Field(default=1000, ge=1)
    """Number of rows deleted per batch during size failsafe enforcement."""

    size_failsafe_vacuum_pages: int = Field(default=100, ge=1)
    """Number of pages to vacuum per size failsafe run."""

    max_consecutive_heartbeat_failures: int = Field(default=3, ge=1)
    """Maximum consecutive heartbeat failures before the database service is considered unhealthy."""

    read_timeout_seconds: float = Field(default=10.0, ge=0.1)
    """Maximum seconds to wait for a telemetry read query before raising TimeoutError."""

    max_flush_interval_seconds: float = Field(default=5.0, ge=0.1)
    """Maximum seconds a record may sit in the CommandExecutor write queue before a
    time-based flush is forced, even if the batch size threshold has not been reached."""

path: Path | None = Field(default=None) class-attribute instance-attribute

Path to the SQLite database file. Defaults to data_dir / "hassette.db" when None.

retention_days: int = Field(default=7, ge=1) class-attribute instance-attribute

Number of days to retain execution records in the executions table.

max_size_mb: float = Field(default=500, ge=0) class-attribute instance-attribute

Maximum database file size in MB. When exceeded, oldest execution records are deleted. 0 disables the size failsafe.

migration_timeout_seconds: int = Field(default=120, ge=10) class-attribute instance-attribute

Maximum seconds to wait for SQL schema migrations to complete at startup.

write_queue_max: int = Field(default=2000, ge=1) class-attribute instance-attribute

Maximum pending coroutines in the DatabaseService write queue. Bounds memory growth under sustained I/O pressure. Fire-and-forget tasks are dropped on overflow; submit() callers block until space is available.

telemetry_write_queue_max: int = Field(default=1000, ge=1) class-attribute instance-attribute

Maximum pending records in the CommandExecutor write queue before records are dropped.

heartbeat_interval_seconds: int = Field(default=300, ge=10) class-attribute instance-attribute

Interval in seconds between database heartbeat checks.

retention_interval_seconds: int = Field(default=3600, ge=60) class-attribute instance-attribute

Interval in seconds between retention enforcement runs.

size_failsafe_interval_seconds: int = Field(default=3600, ge=60) class-attribute instance-attribute

Interval in seconds between size failsafe enforcement runs.

size_failsafe_max_iterations: int = Field(default=10, ge=1) class-attribute instance-attribute

Maximum number of delete-batch iterations per size failsafe run.

size_failsafe_delete_batch: int = Field(default=1000, ge=1) class-attribute instance-attribute

Number of rows deleted per batch during size failsafe enforcement.

size_failsafe_vacuum_pages: int = Field(default=100, ge=1) class-attribute instance-attribute

Number of pages to vacuum per size failsafe run.

max_consecutive_heartbeat_failures: int = Field(default=3, ge=1) class-attribute instance-attribute

Maximum consecutive heartbeat failures before the database service is considered unhealthy.

read_timeout_seconds: float = Field(default=10.0, ge=0.1) class-attribute instance-attribute

Maximum seconds to wait for a telemetry read query before raising TimeoutError.

max_flush_interval_seconds: float = Field(default=5.0, ge=0.1) class-attribute instance-attribute

Maximum seconds a record may sit in the CommandExecutor write queue before a time-based flush is forced, even if the batch size threshold has not been reached.

FileWatcherConfig

Bases: ExcludeExtrasMixin, BaseModel

File watcher debounce, step, and enable/disable settings.

Source code in src/hassette/config/models.py

class FileWatcherConfig(ExcludeExtrasMixin, BaseModel):
    """File watcher debounce, step, and enable/disable settings."""

    debounce_milliseconds: int = Field(default=3000)
    """Debounce time for file watcher events in milliseconds."""

    step_milliseconds: int = Field(default=500)
    """Time to wait for additional file changes before emitting event in milliseconds."""

    watch_files: bool = Field(default=True)
    """Whether to watch files for changes and reload apps automatically."""

debounce_milliseconds: int = Field(default=3000) class-attribute instance-attribute

Debounce time for file watcher events in milliseconds.

step_milliseconds: int = Field(default=500) class-attribute instance-attribute

Time to wait for additional file changes before emitting event in milliseconds.

watch_files: bool = Field(default=True) class-attribute instance-attribute

Whether to watch files for changes and reload apps automatically.

LifecycleConfig

Bases: ExcludeExtrasMixin, BaseModel

Startup, shutdown, and per-operation timeout settings for the resource lifecycle.

Source code in src/hassette/config/models.py

class LifecycleConfig(ExcludeExtrasMixin, BaseModel):
    """Startup, shutdown, and per-operation timeout settings for the resource lifecycle."""

    startup_timeout_seconds: int = Field(default=30)
    """Length of time to wait for each wave of Hassette resources to start before giving up.
    Must be >= app_startup_timeout_seconds since AppHandler readiness waits for app bootstrap."""

    app_startup_timeout_seconds: int = Field(default=20)
    """Length of time to wait for an app to start before giving up."""

    app_shutdown_timeout_seconds: int = Field(default=APP_SHUTDOWN_TIMEOUT_SECONDS)
    """Length of time to wait for an app to shut down before giving up."""

    resource_shutdown_timeout_seconds: int = Field(
        default_factory=lambda data: data.get("app_shutdown_timeout_seconds", APP_SHUTDOWN_TIMEOUT_SECONDS)
    )
    """Per-phase timeout for resource shutdown. Defaults to app_shutdown_timeout_seconds."""

    total_shutdown_timeout_seconds: int = Field(default=30)
    """Maximum wall-clock seconds for the entire Hassette shutdown (hooks + propagation)."""

    registration_await_timeout: int = Field(default=30)
    """Timeout in seconds to wait for all pending listener/job DB registrations to flush
    before post-ready reconciliation. Prevents indefinite hangs if the DB write queue stalls."""

    event_handler_timeout_seconds: float | None = Field(default=600.0)
    """Default timeout in seconds for event handler execution. ``None`` disables the default timeout.
    Individual listeners can override via ``timeout=`` or ``timeout_disabled=True``."""

    error_handler_timeout_seconds: float | None = Field(default=5.0)
    """Default timeout in seconds for error handler execution. ``None`` disables the default timeout."""

    run_sync_timeout_seconds: int | float = Field(default=6)
    """Default timeout for synchronous function calls."""

    task_cancellation_timeout_seconds: int | float = Field(default=5)
    """Length of time to wait for tasks to cancel before forcing."""

    @field_validator("event_handler_timeout_seconds", "error_handler_timeout_seconds", mode="before")
    @classmethod
    def validate_timeouts(cls, value: Any) -> float | None:
        return validate_positive_or_none(value)

startup_timeout_seconds: int = Field(default=30) class-attribute instance-attribute

Length of time to wait for each wave of Hassette resources to start before giving up. Must be >= app_startup_timeout_seconds since AppHandler readiness waits for app bootstrap.

app_startup_timeout_seconds: int = Field(default=20) class-attribute instance-attribute

Length of time to wait for an app to start before giving up.

app_shutdown_timeout_seconds: int = Field(default=APP_SHUTDOWN_TIMEOUT_SECONDS) class-attribute instance-attribute

Length of time to wait for an app to shut down before giving up.

resource_shutdown_timeout_seconds: int = Field(default_factory=(lambda data: data.get('app_shutdown_timeout_seconds', APP_SHUTDOWN_TIMEOUT_SECONDS))) class-attribute instance-attribute

Per-phase timeout for resource shutdown. Defaults to app_shutdown_timeout_seconds.

total_shutdown_timeout_seconds: int = Field(default=30) class-attribute instance-attribute

Maximum wall-clock seconds for the entire Hassette shutdown (hooks + propagation).

registration_await_timeout: int = Field(default=30) class-attribute instance-attribute

Timeout in seconds to wait for all pending listener/job DB registrations to flush before post-ready reconciliation. Prevents indefinite hangs if the DB write queue stalls.

event_handler_timeout_seconds: float | None = Field(default=600.0) class-attribute instance-attribute

Default timeout in seconds for event handler execution. None disables the default timeout. Individual listeners can override via timeout= or timeout_disabled=True.

error_handler_timeout_seconds: float | None = Field(default=5.0) class-attribute instance-attribute

Default timeout in seconds for error handler execution. None disables the default timeout.

run_sync_timeout_seconds: int | float = Field(default=6) class-attribute instance-attribute

Default timeout for synchronous function calls.

task_cancellation_timeout_seconds: int | float = Field(default=5) class-attribute instance-attribute

Length of time to wait for tasks to cancel before forcing.

LoggingConfig

Bases: ExcludeExtrasMixin, BaseModel

Logging level, format, queue, persistence, and per-service log-level settings.

Source code in src/hassette/config/models.py

class LoggingConfig(ExcludeExtrasMixin, BaseModel):
    """Logging level, format, queue, persistence, and per-service log-level settings."""

    log_level: LOG_ANNOTATION = Field(default="INFO")
    """Logging level for Hassette."""

    log_format: Literal["auto", "console", "json"] = Field(default="auto")
    """Console output format. ``"auto"`` detects TTY vs pipe automatically. ``"console"`` forces
    colored human-readable output. ``"json"`` forces one-JSON-object-per-line output."""

    log_queue_max: int = Field(default=2000, ge=1)
    """Maximum size of the inter-thread log queue. Records are dropped when the queue is full."""

    log_persistence_level: LOG_ANNOTATION = Field(default="INFO")
    """Minimum log level for database persistence. Records below this level are not stored."""

    log_retention_days: int = Field(default=3, ge=1)
    """Number of days to retain persisted log records. Must be <= database.retention_days."""

    database_service: LOG_ANNOTATION = Field(default_factory=log_level_default_factory)
    """Logging level for the database service. Defaults to log_level."""

    bus_service: LOG_ANNOTATION = Field(default_factory=log_level_default_factory)
    """Logging level for the event bus service. Defaults to log_level."""

    scheduler_service: LOG_ANNOTATION = Field(default_factory=log_level_default_factory)
    """Logging level for the scheduler service. Defaults to log_level."""

    app_handler: LOG_ANNOTATION = Field(default_factory=log_level_default_factory)
    """Logging level for the app handler service. Defaults to log_level."""

    web_api: LOG_ANNOTATION = Field(default_factory=log_level_default_factory)
    """Logging level for the web API service. Defaults to log_level."""

    websocket: LOG_ANNOTATION = Field(default_factory=log_level_default_factory)
    """Logging level for the WebSocket service. Defaults to log_level."""

    service_watcher: LOG_ANNOTATION = Field(default_factory=log_level_default_factory)
    """Logging level for the service watcher. Defaults to log_level."""

    file_watcher: LOG_ANNOTATION = Field(default_factory=log_level_default_factory)
    """Logging level for the file watcher service. Defaults to log_level."""

    task_bucket: LOG_ANNOTATION = Field(default_factory=log_level_default_factory)
    """Logging level for task buckets. Defaults to log_level."""

    command_executor: LOG_ANNOTATION = Field(default_factory=log_level_default_factory)
    """Logging level for the command executor service. Defaults to log_level."""

    apps: LOG_ANNOTATION = Field(default_factory=log_level_default_factory)
    """Default logging level for apps, can be overridden in app initialization. Defaults to log_level."""

    state_proxy: LOG_ANNOTATION = Field(default_factory=log_level_default_factory)
    """Logging level for the state proxy resource. Defaults to log_level."""

    api: LOG_ANNOTATION = Field(default_factory=log_level_default_factory)
    """Logging level for the API resource (REST/WebSocket client). Defaults to log_level."""

    all_events: bool = Field(default=False)
    """Whether to include all events in bus debug logging. Should be used sparingly."""

    all_hass_events: bool | None = Field(default=None)
    """Whether to include all Home Assistant events in bus debug logging.
    Defaults to False or the value of all_events."""

    all_hassette_events: bool | None = Field(default=None)
    """Whether to include all Hassette events in bus debug logging.
    Defaults to False or the value of all_events."""

    @model_validator(mode="after")
    def fill_event_defaults(self) -> "LoggingConfig":
        """Fill all_hass/hassette_events from all_events when not explicitly set."""
        if self.all_hass_events is None:
            self.all_hass_events = self.all_events
        if self.all_hassette_events is None:
            self.all_hassette_events = self.all_events
        return self

log_level: LOG_ANNOTATION = Field(default='INFO') class-attribute instance-attribute

Logging level for Hassette.

log_format: Literal['auto', 'console', 'json'] = Field(default='auto') class-attribute instance-attribute

Console output format. "auto" detects TTY vs pipe automatically. "console" forces colored human-readable output. "json" forces one-JSON-object-per-line output.

log_queue_max: int = Field(default=2000, ge=1) class-attribute instance-attribute

Maximum size of the inter-thread log queue. Records are dropped when the queue is full.

log_persistence_level: LOG_ANNOTATION = Field(default='INFO') class-attribute instance-attribute

Minimum log level for database persistence. Records below this level are not stored.

log_retention_days: int = Field(default=3, ge=1) class-attribute instance-attribute

Number of days to retain persisted log records. Must be <= database.retention_days.

database_service: LOG_ANNOTATION = Field(default_factory=log_level_default_factory) class-attribute instance-attribute

Logging level for the database service. Defaults to log_level.

bus_service: LOG_ANNOTATION = Field(default_factory=log_level_default_factory) class-attribute instance-attribute

Logging level for the event bus service. Defaults to log_level.

scheduler_service: LOG_ANNOTATION = Field(default_factory=log_level_default_factory) class-attribute instance-attribute

Logging level for the scheduler service. Defaults to log_level.

app_handler: LOG_ANNOTATION = Field(default_factory=log_level_default_factory) class-attribute instance-attribute

Logging level for the app handler service. Defaults to log_level.

web_api: LOG_ANNOTATION = Field(default_factory=log_level_default_factory) class-attribute instance-attribute

Logging level for the web API service. Defaults to log_level.

websocket: LOG_ANNOTATION = Field(default_factory=log_level_default_factory) class-attribute instance-attribute

Logging level for the WebSocket service. Defaults to log_level.

service_watcher: LOG_ANNOTATION = Field(default_factory=log_level_default_factory) class-attribute instance-attribute

Logging level for the service watcher. Defaults to log_level.

file_watcher: LOG_ANNOTATION = Field(default_factory=log_level_default_factory) class-attribute instance-attribute

Logging level for the file watcher service. Defaults to log_level.

task_bucket: LOG_ANNOTATION = Field(default_factory=log_level_default_factory) class-attribute instance-attribute

Logging level for task buckets. Defaults to log_level.

command_executor: LOG_ANNOTATION = Field(default_factory=log_level_default_factory) class-attribute instance-attribute

Logging level for the command executor service. Defaults to log_level.

apps: LOG_ANNOTATION = Field(default_factory=log_level_default_factory) class-attribute instance-attribute

Default logging level for apps, can be overridden in app initialization. Defaults to log_level.

state_proxy: LOG_ANNOTATION = Field(default_factory=log_level_default_factory) class-attribute instance-attribute

Logging level for the state proxy resource. Defaults to log_level.

api: LOG_ANNOTATION = Field(default_factory=log_level_default_factory) class-attribute instance-attribute

Logging level for the API resource (REST/WebSocket client). Defaults to log_level.

all_events: bool = Field(default=False) class-attribute instance-attribute

Whether to include all events in bus debug logging. Should be used sparingly.

all_hass_events: bool | None = Field(default=None) class-attribute instance-attribute

Whether to include all Home Assistant events in bus debug logging. Defaults to False or the value of all_events.

all_hassette_events: bool | None = Field(default=None) class-attribute instance-attribute

Whether to include all Hassette events in bus debug logging. Defaults to False or the value of all_events.

fill_event_defaults() -> LoggingConfig

Fill all_hass/hassette_events from all_events when not explicitly set.

Source code in src/hassette/config/models.py

@model_validator(mode="after")
def fill_event_defaults(self) -> "LoggingConfig":
    """Fill all_hass/hassette_events from all_events when not explicitly set."""
    if self.all_hass_events is None:
        self.all_hass_events = self.all_events
    if self.all_hassette_events is None:
        self.all_hassette_events = self.all_events
    return self

SchedulerConfig

Bases: ExcludeExtrasMixin, BaseModel

Scheduler delay, threshold, and job-timeout settings.

Source code in src/hassette/config/models.py

class SchedulerConfig(ExcludeExtrasMixin, BaseModel):
    """Scheduler delay, threshold, and job-timeout settings."""

    min_delay_seconds: int | float = Field(default=1)
    """Minimum delay between scheduled jobs."""

    max_delay_seconds: int | float = Field(default=30)
    """Maximum delay between scheduled jobs."""

    default_delay_seconds: int | float = Field(default=15)
    """Default delay between scheduled jobs."""

    behind_schedule_threshold_seconds: int | float = Field(default=5)
    """Threshold in seconds before a 'behind schedule' warning is logged for a job."""

    job_timeout_seconds: float | None = Field(default=600.0)
    """Default timeout in seconds for scheduled job execution. ``None`` disables the default timeout.
    Individual jobs can override via ``timeout=`` or ``timeout_disabled=True``."""

    @field_validator("job_timeout_seconds", mode="before")
    @classmethod
    def validate_timeouts(cls, value: Any) -> float | None:
        return validate_positive_or_none(value)

min_delay_seconds: int | float = Field(default=1) class-attribute instance-attribute

Minimum delay between scheduled jobs.

max_delay_seconds: int | float = Field(default=30) class-attribute instance-attribute

Maximum delay between scheduled jobs.

default_delay_seconds: int | float = Field(default=15) class-attribute instance-attribute

Default delay between scheduled jobs.

behind_schedule_threshold_seconds: int | float = Field(default=5) class-attribute instance-attribute

Threshold in seconds before a 'behind schedule' warning is logged for a job.

job_timeout_seconds: float | None = Field(default=600.0) class-attribute instance-attribute

Default timeout in seconds for scheduled job execution. None disables the default timeout. Individual jobs can override via timeout= or timeout_disabled=True.

WebApiConfig

Bases: ExcludeExtrasMixin, BaseModel

Web API and UI server host, port, buffer, and feature-flag settings.

Source code in src/hassette/config/models.py

class WebApiConfig(ExcludeExtrasMixin, BaseModel):
    """Web API and UI server host, port, buffer, and feature-flag settings."""

    run: bool = Field(default=True)
    """Whether to run the web API service (includes healthcheck and UI backend)."""

    run_ui: bool = Field(default=True)
    """Whether to serve the web UI dashboard. Only used when run is True."""

    ui_hot_reload: bool = Field(default=False)
    """Watch web UI static files and templates for changes and push live reloads to the browser."""

    host: str = Field(default="0.0.0.0")
    """Host to bind the web API server to."""

    port: int = Field(default=DEFAULT_WEB_API_PORT)
    """Port to run the web API server on."""

    cors_origins: tuple[str, ...] = Field(default=("http://localhost:3000", "http://localhost:5173"))
    """Allowed CORS origins for the web API, typically the UI dev server."""

    event_buffer_size: int = Field(default=500)
    """Maximum number of recent events to keep in the RuntimeQueryService ring buffer."""

    log_buffer_size: int = Field(default=2000)
    """Maximum number of log entries to keep in the LogCaptureHandler ring buffer."""

    job_history_size: int = Field(default=1000)
    """Maximum number of job execution records to keep."""

run: bool = Field(default=True) class-attribute instance-attribute

Whether to run the web API service (includes healthcheck and UI backend).

run_ui: bool = Field(default=True) class-attribute instance-attribute

Whether to serve the web UI dashboard. Only used when run is True.

ui_hot_reload: bool = Field(default=False) class-attribute instance-attribute

Watch web UI static files and templates for changes and push live reloads to the browser.

host: str = Field(default='0.0.0.0') class-attribute instance-attribute

Host to bind the web API server to.

port: int = Field(default=DEFAULT_WEB_API_PORT) class-attribute instance-attribute

Port to run the web API server on.

cors_origins: tuple[str, ...] = Field(default=('http://localhost:3000', 'http://localhost:5173')) class-attribute instance-attribute

Allowed CORS origins for the web API, typically the UI dev server.

event_buffer_size: int = Field(default=500) class-attribute instance-attribute

Maximum number of recent events to keep in the RuntimeQueryService ring buffer.

log_buffer_size: int = Field(default=2000) class-attribute instance-attribute

Maximum number of log entries to keep in the LogCaptureHandler ring buffer.

job_history_size: int = Field(default=1000) class-attribute instance-attribute

Maximum number of job execution records to keep.

WebSocketConfig

Bases: ExcludeExtrasMixin, BaseModel

WebSocket connection, retry, and recovery timing settings.

Source code in src/hassette/config/models.py

class WebSocketConfig(ExcludeExtrasMixin, BaseModel):
    """WebSocket connection, retry, and recovery timing settings."""

    authentication_timeout_seconds: int = Field(default=10)
    """Length of time to wait for WebSocket authentication to complete."""

    response_timeout_seconds: int = Field(default=15)
    """Length of time to wait for a response from the WebSocket."""

    connection_timeout_seconds: int = Field(default=5)
    """Length of time to wait for WebSocket connection to complete. Passed to aiohttp."""

    total_timeout_seconds: int = Field(default=30)
    """Total length of time to wait for WebSocket operations to complete. Passed to aiohttp."""

    heartbeat_interval_seconds: int = Field(default=30)
    """Interval to send ping messages to keep the WebSocket connection alive. Passed to aiohttp."""

    connect_retry_max_attempts: int = Field(default=5)
    """Maximum number of attempts to establish the initial WebSocket connection before giving up."""

    connect_retry_initial_wait_seconds: float = Field(default=1.0)
    """Initial backoff wait in seconds between WebSocket connection retry attempts."""

    connect_retry_max_wait_seconds: float = Field(default=32.0)
    """Maximum backoff wait in seconds between WebSocket connection retry attempts."""

    early_drop_stable_window_seconds: float = Field(default=30.0)
    """Seconds a connection must stay alive before it is considered stable (resets early-drop counter)."""

    early_drop_max_retries: int = Field(default=5)
    """Maximum number of early-drop reconnect attempts before treating the failure as fatal."""

    early_drop_backoff_initial_seconds: float = Field(default=2.0)
    """Initial backoff wait in seconds between early-drop reconnect attempts."""

    early_drop_backoff_max_seconds: float = Field(default=60.0)
    """Maximum backoff wait in seconds between early-drop reconnect attempts."""

    max_recovery_seconds: float = Field(default=300.0)
    """Maximum total wall-clock seconds to spend on all WebSocket recovery attempts before giving up."""

authentication_timeout_seconds: int = Field(default=10) class-attribute instance-attribute

Length of time to wait for WebSocket authentication to complete.

response_timeout_seconds: int = Field(default=15) class-attribute instance-attribute

Length of time to wait for a response from the WebSocket.

connection_timeout_seconds: int = Field(default=5) class-attribute instance-attribute

Length of time to wait for WebSocket connection to complete. Passed to aiohttp.

total_timeout_seconds: int = Field(default=30) class-attribute instance-attribute

Total length of time to wait for WebSocket operations to complete. Passed to aiohttp.

heartbeat_interval_seconds: int = Field(default=30) class-attribute instance-attribute

Interval to send ping messages to keep the WebSocket connection alive. Passed to aiohttp.

connect_retry_max_attempts: int = Field(default=5) class-attribute instance-attribute

Maximum number of attempts to establish the initial WebSocket connection before giving up.

connect_retry_initial_wait_seconds: float = Field(default=1.0) class-attribute instance-attribute

Initial backoff wait in seconds between WebSocket connection retry attempts.

connect_retry_max_wait_seconds: float = Field(default=32.0) class-attribute instance-attribute

Maximum backoff wait in seconds between WebSocket connection retry attempts.

early_drop_stable_window_seconds: float = Field(default=30.0) class-attribute instance-attribute

Seconds a connection must stay alive before it is considered stable (resets early-drop counter).

early_drop_max_retries: int = Field(default=5) class-attribute instance-attribute

Maximum number of early-drop reconnect attempts before treating the failure as fatal.

early_drop_backoff_initial_seconds: float = Field(default=2.0) class-attribute instance-attribute

Initial backoff wait in seconds between early-drop reconnect attempts.

early_drop_backoff_max_seconds: float = Field(default=60.0) class-attribute instance-attribute

Maximum backoff wait in seconds between early-drop reconnect attempts.

max_recovery_seconds: float = Field(default=300.0) class-attribute instance-attribute

Maximum total wall-clock seconds to spend on all WebSocket recovery attempts before giving up.