Index

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

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

TERMINAL_STATUSES: frozenset[ResourceStatus] = frozenset({ResourceStatus.STOPPED}) 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__.scheduler_service').

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', 'FalseySentinel | V | Condition[V | FalseySentinel] | ComparisonCondition[V | FalseySentinel] | None', type_params=(V,)) module-attribute

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

EventT = TypeVar('EventT', bound='AnyEvent', 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.

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.

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.

BlockingIOBehavior

Bases: StrEnum

Controls what happens when a blocking call is detected on the shared event loop.

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

class BlockingIOBehavior(StrEnum):
    """Controls what happens when a blocking call is detected on the shared event loop."""

    IGNORE = auto()
    """Suppress detection entirely — no warning AND no ``blocking_events`` row is written.
    This is the deliberate exception to the persist-before-warn rule: ``IGNORE`` means
    "I know this app blocks; treat it as if detection never fired," so there is no audit
    trail for ignored events."""

    WARN = auto()
    """Emit a ``HassetteBlockingIOWarning`` (default). Integrates with ``-W error``."""

    ERROR = auto()
    """Emit ``HassetteBlockingIOWarning`` in a form that ``filterwarnings("error")`` escalates
    to a raised exception. Under normal filters, behaves identically to ``WARN`` — the framework
    does not raise on its own; escalation requires the caller to install the error filter
    (``-W error``, a pytest ``filterwarnings`` marker, or ``warnings.filterwarnings`` at startup).

    The effect of escalation differs by detection tier:

    - **Tier 2 (call-site interception)** detects *before* the blocking primitive runs, so an
      escalated warning raises at the call site and prevents the blocking call from executing.
    - **Tier 1 (loop watchdog)** is warn-after: the stall has already completed when the warning
      fires on the off-loop daemon thread. The daemon suppresses the escalation to stay alive, so
      ``ERROR`` cannot retroactively interrupt a Tier 1 stall — it only records and warns."""

IGNORE = auto() class-attribute instance-attribute

Suppress detection entirely — no warning AND no blocking_events row is written. This is the deliberate exception to the persist-before-warn rule: IGNORE means "I know this app blocks; treat it as if detection never fired," so there is no audit trail for ignored events.

WARN = auto() class-attribute instance-attribute

Emit a HassetteBlockingIOWarning (default). Integrates with -W error.

ERROR = auto() class-attribute instance-attribute

Emit HassetteBlockingIOWarning in a form that filterwarnings("error") escalates to a raised exception. Under normal filters, behaves identically to WARN — the framework does not raise on its own; escalation requires the caller to install the error filter (-W error, a pytest filterwarnings marker, or warnings.filterwarnings at startup).

The effect of escalation differs by detection tier:

• Tier 2 (call-site interception) detects before the blocking primitive runs, so an escalated warning raises at the call site and prevents the blocking call from executing.
• Tier 1 (loop watchdog) is warn-after: the stall has already completed when the warning fires on the off-loop daemon thread. The daemon suppresses the escalation to stay alive, so ERROR cannot retroactively interrupt a Tier 1 stall — it only records and warns.

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

    AUTH_FAILED = auto()
    """Home Assistant rejected the access token; Hassette stays up but no longer reconnects.

    Leaving this state takes a process restart, after fixing the token.
    """

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.

AUTH_FAILED = auto() class-attribute instance-attribute

Home Assistant rejected the access token; Hassette stays up but no longer reconnects.

Leaving this state takes a process restart, after fixing the token.

ForgottenAwaitBehavior

Bases: StrEnum

Controls what happens when a protected method is called without await.

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

class ForgottenAwaitBehavior(StrEnum):
    """Controls what happens when a protected method is called without ``await``."""

    IGNORE = auto()
    """Suppress the warning entirely — the forgotten await is silently ignored."""

    WARN = auto()
    """Emit a ``HassetteForgottenAwaitWarning`` (default). Integrates with ``-W error``."""

    ERROR = auto()
    """Emit ``HassetteForgottenAwaitWarning`` in a form that ``filterwarnings("error")`` escalates
    to a raised exception. Under normal filters, behaves identically to ``WARN``."""

IGNORE = auto() class-attribute instance-attribute

Suppress the warning entirely — the forgotten await is silently ignored.

WARN = auto() class-attribute instance-attribute

Emit a HassetteForgottenAwaitWarning (default). Integrates with -W error.

ERROR = auto() class-attribute instance-attribute

Emit HassetteForgottenAwaitWarning in a form that filterwarnings("error") escalates to a raised exception. Under normal filters, behaves identically to WARN.

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

The kind of framework component a resource is. Internal to the server; not part of the wire contract.

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

class ResourceRole(StrEnum):
    """The kind of framework component a resource is. Internal to the server; not part of the wire contract."""

    SERVICE = "service"
    """A long-running framework service with a ``serve()`` task, such as the Home Assistant connection or
    the database. A required one's failure ends the process."""

    RESOURCE = "resource"
    """A framework component that runs for the life of its owner, with no ``serve()`` task of its own."""

    APP = "app"
    """A user app instance."""

SERVICE = 'service' class-attribute instance-attribute

A long-running framework service with a serve() task, such as the Home Assistant connection or the database. A required one's failure ends the process.

RESOURCE = 'resource' class-attribute instance-attribute

A framework component that runs for the life of its owner, with no serve() task of its own.

APP = 'app' class-attribute instance-attribute

A user app instance.

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]:
        """Return an awaitable that runs the handler."""
        ...

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

Return an awaitable that runs the handler.

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

def __call__(self, *args: Any, **kwargs: Any) -> Awaitable[None]:
    """Return an awaitable that runs the handler."""
    ...

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:
        """Return True when the change from old_value to new_value satisfies this comparison."""
        ...

__call__(old_value: V_contra, new_value: V_contra) -> bool

Return True when the change from old_value to new_value satisfies this comparison.

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

def __call__(self, old_value: V_contra, new_value: V_contra, /) -> bool:
    """Return True when the change from old_value to new_value satisfies this comparison."""
    ...

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:
        """Return True when the event satisfies this predicate."""
        ...

__call__(value: EventT) -> bool

Return True when the event satisfies this predicate.

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

def __call__(self, value: EventT, /) -> bool:
    """Return True when the event satisfies this predicate."""
    ...

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 its 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 its 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: "Job") -> None:
        """Persist the job, add it to the live registry, and enqueue it on the heap if it is scheduled."""
        ...

    def dequeue_job(self, job: "Job") -> bool:
        """Remove the job from live state now and finish guard release and persistence in the background.

        Returns:
            True if the job was found and removed from the heap, False otherwise.
        """
        ...

    async def reschedule_job(self, job: "Job", result: "ZonedDateTime | _WaitingSentinel") -> None:
        """Move a registered job to a new run time, or into or out of the WAITING state, keeping its registration."""
        ...

    def register_removal_callback(self, owner_id: str, callback: "Callable[[Job], None]") -> None:
        """Set the callback invoked with each removed job owned by owner_id, replacing any existing one."""
        ...

    def deregister_removal_callback(self, owner_id: str) -> None:
        """Drop the removal callback for owner_id; a no-op when none is registered."""
        ...

    async def mark_job_removed(self, db_id: int) -> None:
        """Persist the removal timestamp on the scheduled-job row identified by db_id."""
        ...

    def deregister_job(self, job: "Job") -> None:
        """Remove this exact job object from the live registry, leaving a different job with the same db_id alone."""
        ...

    async def remove_job(self, job: "Job", *, detach_self: bool = False) -> bool:
        """Remove the job from live state and await its guard release and removal persistence.

        Args:
            job: The job to remove.
            detach_self: Let an invocation that removes its own job finish instead of cancelling it.

        Returns:
            True if the job was found and removed from the heap, False otherwise.
        """
        ...

    def remove_jobs(self, jobs: "list[Job]") -> "asyncio.Future[None]":
        """Start removing exactly the given jobs on the service's own task bucket.

        Returns:
            A future to await for the removals; already completed if the task bucket is sealed.
        """
        ...

    def submit_job(self, job: "Job") -> None:
        """Submit one manual run, subject to the job's overlap guard, without waiting or changing its schedule."""
        ...

add_job(job: Job) -> None async

Persist the job, add it to the live registry, and enqueue it on the heap if it is scheduled.

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

async def add_job(self, job: "Job") -> None:
    """Persist the job, add it to the live registry, and enqueue it on the heap if it is scheduled."""
    ...

dequeue_job(job: Job) -> bool

Remove the job from live state now and finish guard release and persistence in the background.

Returns:

Type	Description
bool	True if the job was found and removed from the heap, False otherwise.

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

def dequeue_job(self, job: "Job") -> bool:
    """Remove the job from live state now and finish guard release and persistence in the background.

    Returns:
        True if the job was found and removed from the heap, False otherwise.
    """
    ...

reschedule_job(job: Job, result: ZonedDateTime | _WaitingSentinel) -> None async

Move a registered job to a new run time, or into or out of the WAITING state, keeping its registration.

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

async def reschedule_job(self, job: "Job", result: "ZonedDateTime | _WaitingSentinel") -> None:
    """Move a registered job to a new run time, or into or out of the WAITING state, keeping its registration."""
    ...

register_removal_callback(owner_id: str, callback: Callable[[Job], None]) -> None

Set the callback invoked with each removed job owned by owner_id, replacing any existing one.

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

def register_removal_callback(self, owner_id: str, callback: "Callable[[Job], None]") -> None:
    """Set the callback invoked with each removed job owned by owner_id, replacing any existing one."""
    ...

deregister_removal_callback(owner_id: str) -> None

Drop the removal callback for owner_id; a no-op when none is registered.

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

def deregister_removal_callback(self, owner_id: str) -> None:
    """Drop the removal callback for owner_id; a no-op when none is registered."""
    ...

mark_job_removed(db_id: int) -> None async

Persist the removal timestamp on the scheduled-job row identified by db_id.

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

async def mark_job_removed(self, db_id: int) -> None:
    """Persist the removal timestamp on the scheduled-job row identified by db_id."""
    ...

deregister_job(job: Job) -> None

Remove this exact job object from the live registry, leaving a different job with the same db_id alone.

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

def deregister_job(self, job: "Job") -> None:
    """Remove this exact job object from the live registry, leaving a different job with the same db_id alone."""
    ...

remove_job(job: Job, *, detach_self: bool = False) -> bool async

Remove the job from live state and await its guard release and removal persistence.

Parameters:

Name	Type	Description	Default
job	Job	The job to remove.	required
detach_self	bool	Let an invocation that removes its own job finish instead of cancelling it.	False

Returns:

Type	Description
bool	True if the job was found and removed from the heap, False otherwise.

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

async def remove_job(self, job: "Job", *, detach_self: bool = False) -> bool:
    """Remove the job from live state and await its guard release and removal persistence.

    Args:
        job: The job to remove.
        detach_self: Let an invocation that removes its own job finish instead of cancelling it.

    Returns:
        True if the job was found and removed from the heap, False otherwise.
    """
    ...

remove_jobs(jobs: list[Job]) -> asyncio.Future[None]

Start removing exactly the given jobs on the service's own task bucket.

Returns:

Type	Description
Future[None]	A future to await for the removals; already completed if the task bucket is sealed.

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

def remove_jobs(self, jobs: "list[Job]") -> "asyncio.Future[None]":
    """Start removing exactly the given jobs on the service's own task bucket.

    Returns:
        A future to await for the removals; already completed if the task bucket is sealed.
    """
    ...

submit_job(job: Job) -> None

Submit one manual run, subject to the job's overlap guard, without waiting or changing its schedule.

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

def submit_job(self, job: "Job") -> None:
    """Submit one manual run, subject to the job's overlap guard, without waiting or changing its schedule."""
    ...

StateReader

Bases: Protocol

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

Describes the two 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 two 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":
        """Return the cached raw state dict for entity_id, or None if it is not cached."""
        ...

    def yield_domain_states(self, domain: str) -> "Generator[tuple[str, HassStateDict], Any, None]":
        """Yield (entity_id, raw state dict) pairs for every cached entity in domain."""
        ...

get_state(entity_id: str) -> HassStateDict | None

Return the cached raw state dict for entity_id, or None if it is not cached.

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

def get_state(self, entity_id: str) -> "HassStateDict | None":
    """Return the cached raw state dict for entity_id, or None if it is not cached."""
    ...

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

Yield (entity_id, raw state dict) pairs for every cached entity in domain.

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

def yield_domain_states(self, domain: str) -> "Generator[tuple[str, HassStateDict], Any, None]":
    """Yield (entity_id, raw state dict) pairs for every cached entity in domain."""
    ...

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:
        """Handle the call synchronously and return any result."""
        ...

__call__(*args: Any, **kwargs: Any) -> Any

Handle the call synchronously and return any result.

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

def __call__(self, *args: Any, **kwargs: Any) -> Any:
    """Handle the call synchronously and return any result."""
    ...

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

first_run_time()/next_run_time() additionally accommodate the _WaitingSentinel (hassette.scheduler.triggers.WAITING) that EntityTime returns when its source entity currently reports no usable time. Every built-in non-EntityTime trigger keeps returning its narrower original type — a narrower return type is compatible with this wider protocol declaration.

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

    ``first_run_time()``/``next_run_time()`` additionally accommodate the ``_WaitingSentinel``
    (``hassette.scheduler.triggers.WAITING``) that ``EntityTime`` returns when its source
    entity currently reports no usable time. Every built-in non-``EntityTime`` trigger keeps
    returning its narrower original type — a narrower return type is compatible with this
    wider protocol declaration.
    """

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

    def next_run_time(
        self, previous_run: ZonedDateTime, current_time: ZonedDateTime
    ) -> "ZonedDateTime | _WaitingSentinel | 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 | _WaitingSentinel

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 | _WaitingSentinel":
    """Return the first scheduled run time at or after current_time."""
    ...

next_run_time(previous_run: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime | _WaitingSentinel | 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 | _WaitingSentinel | 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__.scheduler_service') returns the component slug ('scheduler_service'). 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__.scheduler_service'``) returns the
    component slug (``'scheduler_service'``).  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)