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.

BackpressurePolicy

Bases: StrEnum

What a listener does when the global dispatch semaphore is saturated.

The semaphore is shared by all listeners across all apps — saturation means the whole bus is at capacity, not that this listener alone is busy. This is distinct from per-listener rate controls (debounce, throttle, mode), which operate inside the handler invoker after a dispatch slot is acquired.

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

class BackpressurePolicy(StrEnum):
    """What a listener does when the *global* dispatch semaphore is saturated.

    The semaphore is shared by all listeners across all apps — saturation means
    the whole bus is at capacity, not that this listener alone is busy. This is
    distinct from per-listener rate controls (``debounce``, ``throttle``, ``mode``),
    which operate inside the handler invoker after a dispatch slot is acquired.
    """

    BLOCK = auto()
    """Wait for a dispatch slot — the default for all listeners.

    When the global semaphore is saturated, the dispatch loop blocks until a slot
    opens, then runs the handler. No events are lost; the cost is added latency.
    Omitting ``backpressure=`` on a subscription is identical to passing ``BLOCK``.
    """

    DROP_NEWEST = auto()
    """Skip this event when the global semaphore is saturated; never waits.

    The dispatch loop checks the semaphore without acquiring it. If saturated, no
    task is spawned, one drop is recorded on the listener, and the loop moves on.
    Under normal load (semaphore not locked), dispatches identically to ``BLOCK``.

    A ``DROP_NEWEST`` listener may not run at all during a sustained saturation
    period — every event it receives while the bus is full is dropped. Use ``BLOCK``
    for handlers that must run at least once, even under load.
    """

BLOCK = auto() class-attribute instance-attribute

Wait for a dispatch slot — the default for all listeners.

When the global semaphore is saturated, the dispatch loop blocks until a slot opens, then runs the handler. No events are lost; the cost is added latency. Omitting backpressure= on a subscription is identical to passing BLOCK.

DROP_NEWEST = auto() class-attribute instance-attribute

Skip this event when the global semaphore is saturated; never waits.

The dispatch loop checks the semaphore without acquiring it. If saturated, no task is spawned, one drop is recorded on the listener, and the loop moves on. Under normal load (semaphore not locked), dispatches identically to BLOCK.

A DROP_NEWEST listener may not run at all during a sustained saturation period — every event it receives while the bus is full is dropped. Use BLOCK for handlers that must run at least once, even under load.

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.

ExecutionMode

Bases: StrEnum

Overlap behavior for a listener when a trigger fires while a prior invocation still runs.

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

class ExecutionMode(StrEnum):
    """Overlap behavior for a listener when a trigger fires while a prior invocation still runs."""

    SINGLE = auto()
    """Drop the re-fire while a prior invocation is still running."""

    RESTART = auto()
    """Cancel the running invocation and start a new one."""

    QUEUED = auto()
    """Serialize triggers, running them one at a time in arrival order."""

    PARALLEL = auto()
    """Run invocations concurrently with no overlap guard (today's behavior)."""

SINGLE = auto() class-attribute instance-attribute

Drop the re-fire while a prior invocation is still running.

RESTART = auto() class-attribute instance-attribute

Cancel the running invocation and start a new one.

QUEUED = auto() class-attribute instance-attribute

Serialize triggers, running them one at a time in arrival order.

PARALLEL = auto() class-attribute instance-attribute

Run invocations concurrently with no overlap guard (today's behavior).

Outcome

Bases: StrEnum

The result of handing a trigger to an ExecutionModeGuard.

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

class Outcome(StrEnum):
    """The result of handing a trigger to an ``ExecutionModeGuard``."""

    RAN = auto()
    """The invocation was started immediately."""

    QUEUED_ACCEPTED = auto()
    """A ``queued`` trigger was accepted into the pending queue; it will run later."""

    SUPPRESSED = auto()
    """A ``single`` re-fire was dropped because a prior invocation is still running."""

    DROPPED = auto()
    """A ``queued`` trigger was dropped because the pending queue is at its cap."""

RAN = auto() class-attribute instance-attribute

The invocation was started immediately.

QUEUED_ACCEPTED = auto() class-attribute instance-attribute

A queued trigger was accepted into the pending queue; it will run later.

SUPPRESSED = auto() class-attribute instance-attribute

A single re-fire was dropped because a prior invocation is still running.

DROPPED = auto() class-attribute instance-attribute

A queued trigger was dropped because the pending queue is at its cap.

ResourceRole

Bases: StrEnum

Enumeration for resource roles.

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

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

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

    BASE = auto()
    """The base role for all resources."""

    SERVICE = auto()
    """A service resource."""

    RESOURCE = auto()
    """A generic resource."""

    APP = auto()
    """An application resource."""

    UNKNOWN = auto()
    """An unknown or unclassified resource."""

CORE = auto() class-attribute instance-attribute

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

BASE = auto() class-attribute instance-attribute

The base role for all resources.

SERVICE = auto() class-attribute instance-attribute

A service resource.

RESOURCE = auto() class-attribute instance-attribute

A generic resource.

APP = auto() class-attribute instance-attribute

An application resource.

UNKNOWN = auto() 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: ...

SchedulerServiceProtocol

Bases: Protocol

Protocol for the scheduler-service surface consumed by Scheduler.

Describes the surface Scheduler calls on its scheduler_service: the task_bucket attribute plus six async/sync methods. SchedulerService satisfies this protocol structurally — no changes to the concrete class are required.

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

class SchedulerServiceProtocol(Protocol):
    """Protocol for the scheduler-service surface consumed by Scheduler.

    Describes the surface Scheduler calls on its scheduler_service: the
    ``task_bucket`` attribute plus six async/sync methods. SchedulerService
    satisfies this protocol structurally — no changes to the concrete class
    are required.
    """

    task_bucket: "TaskBucket"

    async def add_job(self, job: "ScheduledJob") -> None: ...

    def dequeue_job(self, job: "ScheduledJob") -> bool: ...

    def register_removal_callback(self, owner_id: str, callback: "Callable[[ScheduledJob], None]") -> None: ...

    def deregister_removal_callback(self, owner_id: str) -> None: ...

    async def mark_job_cancelled(self, db_id: int) -> None: ...

    def remove_jobs_by_owner(self, owner: str) -> "asyncio.Task[None]": ...

StateReader

Bases: Protocol

Read-only protocol for the state-proxy surface consumed by StateManager and DomainStates.

Describes the four members state-manager consumers call on the state proxy. StateProxy satisfies this protocol structurally — no changes to the concrete class are required.

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

class StateReader(Protocol):
    """Read-only protocol for the state-proxy surface consumed by StateManager and DomainStates.

    Describes the four members state-manager consumers call on the state proxy.
    StateProxy satisfies this protocol structurally — no changes to the
    concrete class are required.
    """

    def get_state(self, entity_id: str) -> "HassStateDict | None": ...

    def num_domain_states(self, domain: str) -> int: ...

    def yield_domain_states(self, domain: str) -> "Generator[tuple[str, HassStateDict], Any, None]": ...

    def __contains__(self, entity_id: str) -> 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)