Index

ACTIVE_STATUSES: frozenset[ResourceStatus] = frozenset({ResourceStatus.NOT_STARTED, ResourceStatus.STARTING, ResourceStatus.RUNNING}) module-attribute

Resource is in normal lifecycle progression (not failed, stopped, or exhausted).

TERMINAL_STATUSES: frozenset[ResourceStatus] = frozenset({ResourceStatus.STOPPED, ResourceStatus.EXHAUSTED_DEAD}) module-attribute

Resource has reached an end state — shutdown can skip the STOPPING transition.

FRAMEWORK_APP_KEY = '__hassette__' module-attribute

Reserved app_key for framework-internal listeners and jobs.

Referenced in SQL constraints and in Python guards throughout the codebase. All non-SQL usages should reference this constant or use is_framework_key().

FRAMEWORK_APP_KEY_PREFIX = '__hassette__.' module-attribute

Prefix for component-specific framework keys (e.g. '__hassette__.service_watcher').

The trailing dot distinguishes the prefix from the bare sentinel so that '__hassette__other' is never mistakenly treated as a framework key. Use is_framework_key() rather than comparing against this constant directly.

AppConfigT = TypeVar('AppConfigT', bound='AppConfig') module-attribute

Type variable for app configuration classes.

BusErrorHandlerType: TypeAlias = Callable[['BusErrorContext'], Awaitable[None]] | Callable[['BusErrorContext'], None] module-attribute

Type alias for bus error handler callables.

A bus error handler is either an async callable or a sync callable that accepts a :class:~hassette.bus.error_context.BusErrorContext and returns None.

ChangeType = TypeAliasType('ChangeType', 'None | FalseySentinel | V | Condition[V | FalseySentinel] | ComparisonCondition[V | FalseySentinel]', type_params=(V,)) module-attribute

Type representing a value that can be used to specify changes in predicates.

EventT = TypeVar('EventT', bound='Event[Any]', contravariant=True) module-attribute

Represents an event type.

HandlerType = SyncHandler | AsyncHandlerType module-attribute

Type representing all valid handler types (sync or async).

JobCallable: TypeAlias = Callable[..., Awaitable[None]] | Callable[..., Any] module-attribute

Type representing a callable that can be scheduled as a job.

PayloadT = TypeVar('PayloadT', bound='EventPayload[Any]', covariant=True) module-attribute

Represents the payload type of an event.

QuerySourceTier = Literal['app', 'framework', 'all'] module-attribute

Valid source_tier values for query-side filtering. 'all' disables the filter.

SchedulerErrorHandlerType: TypeAlias = Callable[['SchedulerErrorContext'], Awaitable[None]] | Callable[['SchedulerErrorContext'], None] module-attribute

Type alias for scheduler error handler callables.

A scheduler error handler is either an async callable or a sync callable that accepts a :class:~hassette.scheduler.error_context.SchedulerErrorContext and returns None.

ScheduleStartType: TypeAlias = ZonedDateTime | Time | time | TimeDelta | int | float | None module-attribute

Type representing a value that can be used to specify a start time.

SourceTier = Literal['app', 'framework'] module-attribute

Identifies whether a telemetry record originates from a user app or the framework itself.

StateT = TypeVar('StateT', bound='BaseState', covariant=True) module-attribute

Represents a specific state type, e.g., LightState, CoverState, etc.

StateValueT = TypeVar('StateValueT', covariant=True) module-attribute

Represents the type of the state attribute in a State model, e.g. bool for BinarySensorState.

ConnectionState

Bases: StrEnum

Enumeration for WebSocket connection states.

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

class ConnectionState(StrEnum):
    """Enumeration for WebSocket connection states."""

    DISCONNECTED = auto()
    """The WebSocket connection is not established."""

    CONNECTING = auto()
    """The WebSocket connection is being established."""

    CONNECTED = auto()
    """The WebSocket connection is established and active."""

DISCONNECTED = auto() class-attribute instance-attribute

The WebSocket connection is not established.

CONNECTING = auto() class-attribute instance-attribute

The WebSocket connection is being established.

CONNECTED = auto() class-attribute instance-attribute

The WebSocket connection is established and active.

ResourceRole

Bases: StrEnum

Enumeration for resource roles.

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

class ResourceRole(StrEnum):
    """Enumeration for resource roles."""

    CORE = "Core"
    """Only used by Hassette directly, as it does not inherit from Resource."""

    BASE = "Base"
    """The base role for all resources."""

    SERVICE = "Service"
    """A service resource."""

    RESOURCE = "Resource"
    """A generic resource."""

    APP = "App"
    """An application resource."""

    UNKNOWN = "Unknown"
    """An unknown or unclassified resource."""

CORE = 'Core' class-attribute instance-attribute

Only used by Hassette directly, as it does not inherit from Resource.

BASE = 'Base' class-attribute instance-attribute

The base role for all resources.

SERVICE = 'Service' class-attribute instance-attribute

A service resource.

RESOURCE = 'Resource' class-attribute instance-attribute

A generic resource.

APP = 'App' class-attribute instance-attribute

An application resource.

UNKNOWN = 'Unknown' class-attribute instance-attribute

An unknown or unclassified resource.

ResourceStatus

Bases: StrEnum

Enumeration for resource status.

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

class ResourceStatus(StrEnum):
    """Enumeration for resource status."""

    NOT_STARTED = auto()
    """The resource has not been started yet."""

    STARTING = auto()
    """The resource is in the process of starting."""

    RUNNING = auto()
    """The resource is currently running."""

    STOPPING = auto()
    """The resource is in the process of stopping."""

    STOPPED = auto()
    """The resource has been stopped without errors."""

    FAILED = auto()
    """The resource has failed with a recoverable error."""

    CRASHED = auto()
    """The resource has crashed unexpectedly and cannot recover."""

    EXHAUSTED_DEAD = auto()
    """The service's restart budget is exhausted with no further restarts (permanent end state)."""

    EXHAUSTED_COOLING = auto()
    """The service's restart budget is exhausted and a long cooldown is in progress."""

NOT_STARTED = auto() class-attribute instance-attribute

The resource has not been started yet.

STARTING = auto() class-attribute instance-attribute

The resource is in the process of starting.

RUNNING = auto() class-attribute instance-attribute

The resource is currently running.

STOPPING = auto() class-attribute instance-attribute

The resource is in the process of stopping.

STOPPED = auto() class-attribute instance-attribute

The resource has been stopped without errors.

FAILED = auto() class-attribute instance-attribute

The resource has failed with a recoverable error.

CRASHED = auto() class-attribute instance-attribute

The resource has crashed unexpectedly and cannot recover.

EXHAUSTED_DEAD = auto() class-attribute instance-attribute

The service's restart budget is exhausted with no further restarts (permanent end state).

EXHAUSTED_COOLING = auto() class-attribute instance-attribute

The service's restart budget is exhausted and a long cooldown is in progress.

Topic

Bases: StrEnum

Event topic identifiers for the internal pub/sub bus.

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

class Topic(StrEnum):
    """Event topic identifiers for the internal pub/sub bus."""

    # hassette events

    HASSETTE_EVENT_SERVICE_STATUS = "hassette.event.service_status"
    """Service status updates"""

    HASSETTE_EVENT_WEBSOCKET_CONNECTED = "hassette.event.websocket_connected"
    """WebSocket connection established"""

    HASSETTE_EVENT_WEBSOCKET_DISCONNECTED = "hassette.event.websocket_disconnected"
    """WebSocket connection lost"""

    HASSETTE_EVENT_FILE_WATCHER = "hassette.event.file_watcher"
    """File watcher events"""

    HASSETTE_EVENT_APP_LOAD_COMPLETED = "hassette.event.app_load_completed"
    """Application load completion events"""

    HASSETTE_EVENT_APP_STATE_CHANGED = "hassette.event.app_state_changed"
    """App instance state change events"""

    HASSETTE_EVENT_EXECUTION_COMPLETED = "hassette.event.execution_completed"
    """Handler or job execution persisted to the telemetry database"""

    # Home Assistant events

    HASS_EVENT_STATE_CHANGED = "hass.event.state_changed"
    """State change events"""

    HASS_EVENT_CALL_SERVICE = "hass.event.call_service"
    """Service call events"""

    HASS_EVENT_COMPONENT_LOADED = "hass.event.component_loaded"
    """Component loaded events"""

    HASS_EVENT_SERVICE_REGISTERED = "hass.event.service_registered"
    """Service registered events"""

    HASS_EVENT_SERVICE_REMOVED = "hass.event.service_removed"
    """Service removed events"""

    HASS_EVENT_LOGBOOK_ENTRY = "hass.event.logbook_entry"
    """Logbook entry events"""

    HASS_EVENT_USER_ADDED = "hass.event.user_added"
    """User added events"""

    HASS_EVENT_USER_REMOVED = "hass.event.user_removed"
    """User removed events"""

    HASS_EVENT_AUTOMATION_TRIGGERED = "hass.event.automation_triggered"
    """Automation triggered events"""

    HASS_EVENT_SCRIPT_STARTED = "hass.event.script_started"
    """Script started events"""

HASSETTE_EVENT_SERVICE_STATUS = 'hassette.event.service_status' class-attribute instance-attribute

Service status updates

HASSETTE_EVENT_WEBSOCKET_CONNECTED = 'hassette.event.websocket_connected' class-attribute instance-attribute

WebSocket connection established

HASSETTE_EVENT_WEBSOCKET_DISCONNECTED = 'hassette.event.websocket_disconnected' class-attribute instance-attribute

WebSocket connection lost

HASSETTE_EVENT_FILE_WATCHER = 'hassette.event.file_watcher' class-attribute instance-attribute

File watcher events

HASSETTE_EVENT_APP_LOAD_COMPLETED = 'hassette.event.app_load_completed' class-attribute instance-attribute

Application load completion events

HASSETTE_EVENT_APP_STATE_CHANGED = 'hassette.event.app_state_changed' class-attribute instance-attribute

App instance state change events

HASSETTE_EVENT_EXECUTION_COMPLETED = 'hassette.event.execution_completed' class-attribute instance-attribute

Handler or job execution persisted to the telemetry database

HASS_EVENT_STATE_CHANGED = 'hass.event.state_changed' class-attribute instance-attribute

State change events

HASS_EVENT_CALL_SERVICE = 'hass.event.call_service' class-attribute instance-attribute

Service call events

HASS_EVENT_COMPONENT_LOADED = 'hass.event.component_loaded' class-attribute instance-attribute

Component loaded events

HASS_EVENT_SERVICE_REGISTERED = 'hass.event.service_registered' class-attribute instance-attribute

Service registered events

HASS_EVENT_SERVICE_REMOVED = 'hass.event.service_removed' class-attribute instance-attribute

Service removed events

HASS_EVENT_LOGBOOK_ENTRY = 'hass.event.logbook_entry' class-attribute instance-attribute

Logbook entry events

HASS_EVENT_USER_ADDED = 'hass.event.user_added' class-attribute instance-attribute

User added events

HASS_EVENT_USER_REMOVED = 'hass.event.user_removed' class-attribute instance-attribute

User removed events

HASS_EVENT_AUTOMATION_TRIGGERED = 'hass.event.automation_triggered' class-attribute instance-attribute

Automation triggered events

HASS_EVENT_SCRIPT_STARTED = 'hass.event.script_started' class-attribute instance-attribute

Script started events

AsyncHandlerType

Bases: Protocol

Protocol for async handlers.

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

class AsyncHandlerType(Protocol):
    """Protocol for async handlers."""

    def __call__(self, *args: Any, **kwargs: Any) -> Awaitable[None]: ...

ComparisonCondition

Bases: Protocol[V_contra]

Protocol for a comparison condition callable that takes two values and returns a bool.

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

class ComparisonCondition(Protocol[V_contra]):
    """Protocol for a comparison condition callable that takes two values and returns a bool."""

    def __call__(self, old_value: V_contra, new_value: V_contra, /) -> bool: ...

Predicate

Bases: Protocol[EventT]

Protocol for defining predicates that evaluate events.

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

class Predicate(Protocol[EventT]):
    """Protocol for defining predicates that evaluate events."""

    def __call__(self, value: EventT, /) -> bool: ...

SyncHandler

Bases: Protocol

Protocol for sync handlers.

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

class SyncHandler(Protocol):
    """Protocol for sync handlers."""

    def __call__(self, *args: Any, **kwargs: Any) -> Any: ...

TriggerProtocol

Bases: Protocol

Protocol for defining triggers.

Six methods make up the contract: - first_run_time: returns the first scheduled run time given the current time - next_run_time: returns the next run time after a previous run, or None for one-shot triggers - trigger_label: short stable label for telemetry / UI display - trigger_detail: optional human-readable detail string - trigger_db_type: canonical type string for database storage - trigger_id: stable string identifier used for deduplication

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

@runtime_checkable
class TriggerProtocol(Protocol):
    """Protocol for defining triggers.

    Six methods make up the contract:
    - first_run_time: returns the first scheduled run time given the current time
    - next_run_time: returns the next run time after a previous run, or None for one-shot triggers
    - trigger_label: short stable label for telemetry / UI display
    - trigger_detail: optional human-readable detail string
    - trigger_db_type: canonical type string for database storage
    - trigger_id: stable string identifier used for deduplication
    """

    def first_run_time(self, current_time: ZonedDateTime) -> ZonedDateTime:
        """Return the first scheduled run time at or after current_time."""
        ...

    def next_run_time(self, previous_run: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime | None:
        """Return the next run time after previous_run, or None for one-shot triggers."""
        ...

    def trigger_label(self) -> str:
        """Human-readable display label for the UI.

        Custom triggers (those returning ``"custom"`` from ``trigger_db_type()``)
        MUST NOT return one of the built-in reserved names (``"after"``,
        ``"once"``, ``"every"``, ``"daily"``, ``"cron"``) from this method.
        Doing so creates misleading telemetry and UI rows where the
        ``trigger_type`` column is ``"custom"`` but the label implies a
        built-in trigger kind.
        """
        ...

    def trigger_detail(self) -> str | None:
        """Optional human-readable detail string."""
        ...

    def trigger_db_type(self) -> Literal["interval", "cron", "once", "after", "custom"]:
        """Canonical type string for database storage."""
        ...

    def trigger_id(self) -> str:
        """Stable string identifier used for deduplication."""
        ...

first_run_time(current_time: ZonedDateTime) -> ZonedDateTime

Return the first scheduled run time at or after current_time.

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

def first_run_time(self, current_time: ZonedDateTime) -> ZonedDateTime:
    """Return the first scheduled run time at or after current_time."""
    ...

next_run_time(previous_run: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime | None

Return the next run time after previous_run, or None for one-shot triggers.

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

def next_run_time(self, previous_run: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime | None:
    """Return the next run time after previous_run, or None for one-shot triggers."""
    ...

trigger_label() -> str

Human-readable display label for the UI.

Custom triggers (those returning "custom" from trigger_db_type()) MUST NOT return one of the built-in reserved names ("after", "once", "every", "daily", "cron") from this method. Doing so creates misleading telemetry and UI rows where the trigger_type column is "custom" but the label implies a built-in trigger kind.

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

def trigger_label(self) -> str:
    """Human-readable display label for the UI.

    Custom triggers (those returning ``"custom"`` from ``trigger_db_type()``)
    MUST NOT return one of the built-in reserved names (``"after"``,
    ``"once"``, ``"every"``, ``"daily"``, ``"cron"``) from this method.
    Doing so creates misleading telemetry and UI rows where the
    ``trigger_type`` column is ``"custom"`` but the label implies a
    built-in trigger kind.
    """
    ...

trigger_detail() -> str | None

Optional human-readable detail string.

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

def trigger_detail(self) -> str | None:
    """Optional human-readable detail string."""
    ...

trigger_db_type() -> Literal['interval', 'cron', 'once', 'after', 'custom']

Canonical type string for database storage.

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

def trigger_db_type(self) -> Literal["interval", "cron", "once", "after", "custom"]:
    """Canonical type string for database storage."""
    ...

trigger_id() -> str

Stable string identifier used for deduplication.

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

def trigger_id(self) -> str:
    """Stable string identifier used for deduplication."""
    ...

framework_display_name(app_key: str) -> str

Return a human-readable display name for a framework app key.

For prefixed keys (e.g. '__hassette__.service_watcher') returns the component slug ('service_watcher'). For the bare legacy key '__hassette__' returns 'framework'.

Parameters:

Name	Type	Description	Default
app_key	str	A framework app key. Behaviour is undefined for non-framework keys.	required

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

def framework_display_name(app_key: str) -> str:
    """Return a human-readable display name for a framework app key.

    For prefixed keys (e.g. ``'__hassette__.service_watcher'``) returns the
    component slug (``'service_watcher'``).  For the bare legacy key
    ``'__hassette__'`` returns ``'framework'``.

    Args:
        app_key: A framework app key.  Behaviour is undefined for non-framework keys.
    """
    if app_key.startswith(FRAMEWORK_APP_KEY_PREFIX):
        return app_key.removeprefix(FRAMEWORK_APP_KEY_PREFIX)
    return "framework"

is_framework_key(app_key: str | None) -> bool

Return True if app_key is a framework-reserved key.

Matches both the legacy bare key '__hassette__' and any key with the component prefix '__hassette__.'.

Parameters:

Name	Type	Description	Default
app_key	str | None	The app key to test. None returns False.	required

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

def is_framework_key(app_key: str | None) -> bool:
    """Return True if *app_key* is a framework-reserved key.

    Matches both the legacy bare key ``'__hassette__'`` and any key with the
    component prefix ``'__hassette__.'``.

    Args:
        app_key: The app key to test. ``None`` returns ``False``.
    """
    if app_key is None:
        return False
    return app_key == FRAMEWORK_APP_KEY or app_key.startswith(FRAMEWORK_APP_KEY_PREFIX)