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

    autostart: bool = Field(default=True)
    """Whether the app starts automatically when Hassette starts. Orthogonal to
    `enabled`: an enabled app with autostart=false is registered and startable on
    demand, but is not started at startup or by a live config reload."""

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

    cache_key: str = Field(default="")
    """Override the cache directory key. When empty (default), App.cache_key computes
    '{app_key}/{index}'. When set, this value is used as-is."""

    def __repr__(self) -> str:
        return (
            f"<AppManifest {self.display_name} ({self.class_name})"
            f" - enabled={self.enabled} autostart={self.autostart} 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

    @field_validator("cache_key")
    @classmethod
    def validate_cache_key(cls, v: str) -> str:
        if not v:
            return v
        if is_framework_key(v):
            raise ValueError(f"cache_key {v!r} uses a framework-reserved prefix")
        parsed = PurePath(v)
        if parsed.is_absolute():
            raise ValueError(f"cache_key {v!r} must be a relative path")
        if ".." in parsed.parts:
            raise ValueError(f"cache_key {v!r} must not contain parent-directory traversal")
        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.

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

Whether the app starts automatically when Hassette starts. Orthogonal to enabled: an enabled app with autostart=false is registered and startable on demand, but is not started at startup or by a live config reload.

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

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

Override the cache directory key. When empty (default), App.cache_key computes '{app_key}/{index}'. When set, this value is used as-is.

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

    blocking_io: BlockingIODetectionConfig = Field(default_factory=BlockingIODetectionConfig)
    """Blocking-I/O detection settings for the shared event loop."""

    timezone: str | None = Field(default=None)
    """IANA timezone name (e.g. ``"America/Chicago"``) used for all wall-clock
    operations: scheduler trigger resolution, event timestamp conversion, and
    state model datetime fields.

    When ``None`` (default), the system process timezone is used. Set this when
    the hassette process runs in a different timezone than Home Assistant (e.g.
    Docker containers running UTC while HA is configured for a local zone)."""

    # 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", json_schema_extra={"ui": {"label": "Base URL"}})
    """Base URL of the Home Assistant instance"""

    verify_ssl: bool = Field(default=True, json_schema_extra={"ui": {"label": "Verify SSL"}})
    """Whether to verify SSL certificates when connecting to Home Assistant. Useful to disable for self-signed
    certificates."""

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

    Stored as a :class:`~pydantic.SecretStr` so the value is masked in logs
    and string representations.  Unwrap with ``token.get_secret_value()`` when
    the plaintext is required (e.g. HTTP auth headers, WebSocket auth payload).
    """

    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_ttl: int | None = Field(default=None)
    """Default TTL for cache entries in seconds. None means entries persist indefinitely."""

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

    forgotten_await_behavior: ForgottenAwaitBehavior | None = Field(default=None)
    """Global default for forgotten-await detection behavior.

    When ``None`` (default), the effective behavior is ``"warn"``.  Per-app
    ``AppConfig.forgotten_await_behavior`` overrides this when set.  Set to ``"ignore"``
    to suppress warnings globally, or ``"error"`` to escalate via ``filterwarnings("error")``."""

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

        Calls :meth:`~pydantic.SecretStr.get_secret_value` to unwrap the token
        so the plaintext reaches the Authorization header, not the masked repr.
        """
        if self.token is None:
            return {}
        return {"Authorization": f"Bearer {self.token.get_secret_value()}"}

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

        Calls :meth:`~pydantic.SecretStr.get_secret_value` to obtain the
        plaintext before slicing — ``SecretStr`` is not subscriptable.
        """
        if self.token is None:
            return "<not set>"
        raw = self.token.get_secret_value()
        if len(raw) < TOKEN_SHORT_THRESHOLD:
            return "***"
        if len(raw) <= TOKEN_MEDIUM_THRESHOLD:
            return f"{raw[:TOKEN_SHORT_PREFIX_LENGTH]}***"
        return f"{raw[:TOKEN_LONG_PREFIX_LENGTH]}...{raw[-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("timezone", mode="before")
    @classmethod
    def validate_timezone(cls, value: str | None) -> str | None:
        if value is None:
            return None
        try:
            ZoneInfo(value)
        except (KeyError, ValueError):
            raise ValueError(
                f"Invalid timezone: {value!r}. Use an IANA timezone name (e.g. 'America/Chicago', 'Europe/London')."
            ) from None
        return value

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

    @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

        warn_on_cache_key_collisions(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.

blocking_io: BlockingIODetectionConfig = Field(default_factory=BlockingIODetectionConfig) class-attribute instance-attribute

Blocking-I/O detection settings for the shared event loop.

timezone: str | None = Field(default=None) class-attribute instance-attribute

IANA timezone name (e.g. "America/Chicago") used for all wall-clock operations: scheduler trigger resolution, event timestamp conversion, and state model datetime fields.

When None (default), the system process timezone is used. Set this when the hassette process runs in a different timezone than Home Assistant (e.g. Docker containers running UTC while HA is configured for a local zone).

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', json_schema_extra={'ui': {'label': 'Base URL'}}) class-attribute instance-attribute

Base URL of the Home Assistant instance

verify_ssl: bool = Field(default=True, json_schema_extra={'ui': {'label': 'Verify SSL'}}) class-attribute instance-attribute

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

token: SecretStr | 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.

Stored as a :class:~pydantic.SecretStr so the value is masked in logs and string representations. Unwrap with token.get_secret_value() when the plaintext is required (e.g. HTTP auth headers, WebSocket auth payload).

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_ttl: int | None = Field(default=None) class-attribute instance-attribute

Default TTL for cache entries in seconds. None means entries persist indefinitely.

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.

forgotten_await_behavior: ForgottenAwaitBehavior | None = Field(default=None) class-attribute instance-attribute

Global default for forgotten-await detection behavior.

When None (default), the effective behavior is "warn". Per-app AppConfig.forgotten_await_behavior overrides this when set. Set to "ignore" to suppress warnings globally, or "error" to escalate via filterwarnings("error").

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.

Calls :meth:~pydantic.SecretStr.get_secret_value to unwrap the token so the plaintext reaches the Authorization header, not the masked repr.

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.

Calls :meth:~pydantic.SecretStr.get_secret_value to obtain the plaintext before slicing — SecretStr is not subscriptable.

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

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

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

    failed_instance_cleanup_timeout_seconds: int = Field(default=5)
    """Length of time to wait for listener/job cleanup after an app instance fails to initialize."""

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

    sync_executor_max_workers: int = Field(default_factory=lambda: min(32, (os.cpu_count() or 1) + 4), ge=1)
    """Pool ceiling for the dedicated sync-handler thread pool executor.
    Sized to match the prior implicit default-pool ceiling (min(32, cpu+4)). This is a
    reasonable starting ceiling, NOT a literal behavior-equivalence guarantee: the old
    shared pool also served logging and DB work, so a same-size dedicated pool gives sync
    handlers more effective headroom in practice."""

    sync_executor_shutdown_timeout_seconds: float = Field(default=10.0, gt=0)
    """Join-or-interrupt budget for sync-handler worker threads at shutdown (ported from HA).
    Must be less than total_shutdown_timeout_seconds so the outer shutdown budget is not
    exceeded by the executor interrupt phase."""

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

    max_concurrent_dispatches: int = Field(default=50, ge=1)
    """Ceiling on concurrent event-handler invocations dispatched by the bus.

    The bus fans an event out to one task per matching listener. Without a bound, a state
    storm hitting many listeners spawns unbounded tasks and can exhaust memory and the event
    loop. When this many handlers are already in flight, the bus waits for a slot before
    spawning more — that wait propagates back through the inbound event channel to the
    Home Assistant WebSocket reader (true backpressure).

    A running handler holds its slot until it returns or hits ``event_handler_timeout_seconds``,
    so a slow handler can stall dispatch up to that timeout. Read once at startup; changing it
    requires a restart."""

    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)

    @model_validator(mode="after")
    def validate_sync_executor_shutdown_budget(self) -> "LifecycleConfig":
        """Ensure the sync executor shutdown budget stays within the total shutdown window."""
        if self.sync_executor_shutdown_timeout_seconds >= self.total_shutdown_timeout_seconds:
            raise ValueError(
                f"sync_executor_shutdown_timeout_seconds "
                f"({self.sync_executor_shutdown_timeout_seconds}) must be less than "
                f"total_shutdown_timeout_seconds ({self.total_shutdown_timeout_seconds})"
            )
        return self

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.

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

Length of time to wait for listener/job cleanup after an app instance fails to initialize.

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

sync_executor_max_workers: int = Field(default_factory=(lambda: min(32, (os.cpu_count() or 1) + 4)), ge=1) class-attribute instance-attribute

Pool ceiling for the dedicated sync-handler thread pool executor. Sized to match the prior implicit default-pool ceiling (min(32, cpu+4)). This is a reasonable starting ceiling, NOT a literal behavior-equivalence guarantee: the old shared pool also served logging and DB work, so a same-size dedicated pool gives sync handlers more effective headroom in practice.

sync_executor_shutdown_timeout_seconds: float = Field(default=10.0, gt=0) class-attribute instance-attribute

Join-or-interrupt budget for sync-handler worker threads at shutdown (ported from HA). Must be less than total_shutdown_timeout_seconds so the outer shutdown budget is not exceeded by the executor interrupt phase.

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.

max_concurrent_dispatches: int = Field(default=50, ge=1) class-attribute instance-attribute

Ceiling on concurrent event-handler invocations dispatched by the bus.

The bus fans an event out to one task per matching listener. Without a bound, a state storm hitting many listeners spawns unbounded tasks and can exhaust memory and the event loop. When this many handlers are already in flight, the bus waits for a slot before spawning more — that wait propagates back through the inbound event channel to the Home Assistant WebSocket reader (true backpressure).

A running handler holds its slot until it returns or hits event_handler_timeout_seconds, so a slow handler can stall dispatch up to that timeout. Read once at startup; changing it requires a restart.

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.

validate_sync_executor_shutdown_budget() -> LifecycleConfig

Ensure the sync executor shutdown budget stays within the total shutdown window.

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

@model_validator(mode="after")
def validate_sync_executor_shutdown_budget(self) -> "LifecycleConfig":
    """Ensure the sync executor shutdown budget stays within the total shutdown window."""
    if self.sync_executor_shutdown_timeout_seconds >= self.total_shutdown_timeout_seconds:
        raise ValueError(
            f"sync_executor_shutdown_timeout_seconds "
            f"({self.sync_executor_shutdown_timeout_seconds}) must be less than "
            f"total_shutdown_timeout_seconds ({self.total_shutdown_timeout_seconds})"
        )
    return self

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

    # A group label (section title) must live on the model's own `model_config`, NOT on the
    # `Field(json_schema_extra=...)` of the group in HassetteConfig: a nested-model field is
    # emitted as a `$ref`, and the server-side deref (`jsonref.replace_refs`) drops `$ref`
    # sibling keys, so a `ui` block placed there is silently lost. Scalar field labels are not
    # `$ref` nodes and so use `Field(json_schema_extra=...)` as usual.
    model_config = ConfigDict(json_schema_extra={"ui": {"group_label": "Web API"}})

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

    run_ui: bool = Field(default=True, json_schema_extra={"ui": {"label": "Run UI"}})
    """Whether to serve the web UI dashboard. Only used when run is True."""

    ui_hot_reload: bool = Field(default=False, json_schema_extra={"ui": {"label": "UI Hot Reload"}})
    """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"),
        json_schema_extra={"ui": {"label": "CORS Origins"}},
    )
    """Allowed CORS origins for the web API, typically the UI dev server."""

    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, json_schema_extra={'ui': {'label': 'Run UI'}}) class-attribute instance-attribute

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

ui_hot_reload: bool = Field(default=False, json_schema_extra={'ui': {'label': 'UI Hot Reload'}}) 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'), json_schema_extra={'ui': {'label': 'CORS Origins'}}) class-attribute instance-attribute

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

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

    cleanup_timeout_seconds: float = Field(default=2.0)
    """Length of time to wait for the recv task to finish cancelling during partial cleanup."""

    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.

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

Length of time to wait for the recv task to finish cancelling during partial cleanup.

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.