Index

Task scheduling functionality for Home Assistant automations.

This module provides clean access to the scheduler system for running jobs at specific times, intervals, or based on cron expressions.

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

ScheduledJob dataclass

A job scheduled to run based on a trigger or at a specific time.

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

@dataclass(order=True)
class ScheduledJob:
    """A job scheduled to run based on a trigger or at a specific time."""

    sort_index: tuple[int, int] = field(init=False, repr=False)
    """Tuple of (next_run timestamp with nanoseconds, object id) for ordering in a priority queue."""

    owner_id: str = field(compare=False)
    """Unique string identifier for the owner of the job, e.g., a component or integration name."""

    next_run: ZonedDateTime = field(compare=False)
    """Unjittered logical fire time — used as `previous_run` in subsequent trigger calls."""

    fire_at: ZonedDateTime = field(init=False, compare=False)
    """Actual dispatch time, including any jitter offset.

    Equals ``next_run`` when no jitter is configured. Set by
    ``SchedulerService.apply_jitter_to_heap()`` at enqueue time when jitter > 0.
    The pop loop in ``_ScheduledJobQueue.pop_due_and_peek_next`` compares against
    ``fire_at`` (not ``next_run``) to decide when to dispatch.
    """

    job: "JobCallable" = field(compare=False)
    """The callable to execute when the job runs."""

    app_key: str = field(default="", compare=False)
    """Configuration-level app key for DB registration (e.g., 'my_app'). Empty for non-App owners."""

    instance_index: int = field(default=0, compare=False)
    """App instance index for DB registration. 0 for non-App owners."""

    instance_name: str | None = field(default=None, compare=False)
    """App instance name, precomputed at registration time from the owning app's app_config.instance_name.
    None for framework-tier jobs and non-App owners."""

    trigger: "TriggerProtocol | None" = field(compare=False, default=None)
    """The trigger that determines the job's schedule."""

    group: str | None = field(default=None, compare=False)
    """Optional group name for grouping related jobs. Included in deduplication comparison."""

    jitter: float | None = field(default=None, compare=False)
    """Seconds of random offset applied at enqueue time by ``SchedulerService.apply_jitter_to_heap()``.

    Does not affect ``next_run`` (unjittered logical fire time). See the ``fire_at`` field on
    ``ScheduledJob`` for the actual dispatch time after jitter is applied.
    """

    timeout: float | None = field(default=None, compare=False)
    """Per-job timeout in seconds. ``None`` means use the global default
    (``config.scheduler_job_timeout_seconds``). A positive ``float`` overrides the default.
    Validated at construction: must be positive when set."""

    timeout_disabled: bool = field(default=False, compare=False)
    """When ``True``, timeout enforcement is disabled for this job regardless of the global default."""

    name: str = field(default="", compare=False)
    """Name for the job, used for identification. Required by all public Scheduler entry points
    (``Scheduler.add_job()`` and ``Scheduler.schedule()`` raise ``SchedulerNameRequiredError`` if empty)."""

    args: tuple[Any, ...] = field(default_factory=tuple, compare=False)
    """Positional arguments to pass to the job callable."""

    kwargs: dict[str, Any] = field(default_factory=dict, compare=False)
    """Keyword arguments to pass to the job callable."""

    error_handler: "SchedulerErrorHandlerType | None" = field(default=None, compare=False)
    """Optional error handler for this job.

    When set, this handler is invoked if the job raises an exception (including
    ``TimeoutError``, but excluding ``CancelledError``). Stored as-is for identity comparison
    in ``matches()``. ``compare=False`` prevents ``Callable | None`` from corrupting
    the ``@dataclass(order=True)`` heap ordering.
    """

    db_id: int | None = field(default=None, compare=False)
    """Database row ID for this job. Set by the executor after persistence; None until then."""

    source_location: str = field(default="", compare=False)
    """Captured source location (file:line) of the user code that scheduled this job."""

    registration_source: str = field(default="", compare=False)
    """Captured source code snippet of the scheduling call."""

    source_tier: SourceTier = field(default="app", compare=False)
    """Whether this job originates from a user app or the framework itself."""

    mode: ExecutionMode = field(default=ExecutionMode.SINGLE, compare=False)
    """Resolved overlap mode for this job. Determines behavior when a prior invocation is still
    running as the next occurrence becomes due. Defaults to ``ExecutionMode.SINGLE`` so existing
    direct ``ScheduledJob(...)`` constructions in tests remain valid; the real resolution happens
    in ``Scheduler.schedule()``."""

    predicate: "SchedulerPredicate | None" = field(default=None, compare=False)
    """Normalized predicate callable, or ``None`` when no ``where=`` was given.

    Set once at registration time by ``Scheduler.schedule()`` — a single callable is stored
    directly, and a sequence of predicates is collapsed into a combinator. ``compare=False``
    prevents ``Callable | None`` from corrupting the ``@dataclass(order=True)`` heap ordering.
    """

    predicate_invoker: "CallableInvoker | None" = field(default=None, repr=False, compare=False)
    """Pre-built DI invoker for the predicate, or ``None`` when no ``where=`` was given.

    Built once at registration time by ``Scheduler.schedule()`` using the shared DI layer
    (``hassette.di``) and passed alongside ``predicate``. At dispatch time,
    ``predicate_invoker.invoke({ScheduledJob: job})`` produces kwargs and the predicate is
    called with ``predicate(**kwargs)``."""

    guard: ExecutionModeGuard = field(init=False, compare=False)
    """Per-job overlap state machine. Created in ``__post_init__`` from ``mode``. Lives for the
    full lifetime of the job, spanning all re-fires. ``compare=False`` prevents the guard from
    affecting heap ordering. ``init=False`` — do not pass directly; set via ``mode``."""

    _scheduler: "Scheduler | None" = field(default=None, repr=False, compare=False)
    """Back-reference to the Scheduler that owns this job. Set by Scheduler.add_job()."""

    app_error_handler_resolver: "Callable[[], SchedulerErrorHandlerType | None] | None" = field(
        default=None, init=False, repr=False
    )
    """Closure that resolves the app-level error handler at dispatch time."""

    _dequeued: bool = field(default=False, repr=False, compare=False)
    """True after the job has been synchronously removed from the heap via dequeue_job()."""

    pending_done: "set[asyncio.Future[None]]" = field(default_factory=set, init=False, repr=False, compare=False)
    """Unresolved per-invocation completion futures for non-parallel modes.

    Each ``run_job_with_guard`` call for single/restart/queued parks the outer dispatch task on
    a future that resolves when the invocation actually runs (or is dropped/released). A queued
    invocation accepted into the guard deque has no live child until drain time, so its future
    would hang forever if the job is cancelled first. ``dequeue_job``, ``_remove_job``, and
    ``_remove_jobs_by_owner`` call ``drain_pending_done(job.pending_done)`` after
    ``guard.release()`` so those dispatch tasks unwind.
    """

    def __hash__(self) -> int:
        # Hashing on object identity is safe: each ScheduledJob is a unique object,
        # and sort_index includes id(self) as the tiebreaker, so the hash contract
        # (a == b implies hash(a) == hash(b)) holds. @dataclass(order=True) generates
        # __eq__ based on all compare=True fields (sort_index only), so two distinct
        # objects with the same sort_index are unequal and can safely share a hash bucket.
        return id(self)

    def __repr__(self) -> str:
        return f"ScheduledJob(name={self.name!r}, owner_id={self.owner_id})"

    def __post_init__(self) -> None:
        if self.timeout is not None and (isinstance(self.timeout, bool) or self.timeout <= 0):
            raise ValueError("timeout must be a positive number")
        if self.timeout_disabled and self.timeout is not None:
            raise ValueError("Cannot specify both 'timeout' and 'timeout_disabled=True'")

        self.guard = ExecutionModeGuard(self.mode)
        self.set_next_run(self.next_run)

        self.args = tuple(self.args)
        self.kwargs = dict(self.kwargs)

    def mark_registered(self, db_id: int) -> None:
        """Set the database ID. Called by SchedulerService.add_job() after persistence.

        First call wins — a second call keeps the original id and logs a WARNING so a
        retry or double-registration is surfaced rather than silently swallowed.
        Mirrors ``Listener.mark_registered``.
        """
        if self.db_id is not None:
            LOGGER.warning(
                "ScheduledJob %s already registered with db_id=%s, ignoring new db_id=%s",
                self.name,
                self.db_id,
                db_id,
            )
            return
        self.db_id = db_id

    def matches(self, other: "ScheduledJob") -> bool:
        """Check whether two jobs represent the same logical configuration.

        Compares callable, trigger (by trigger_id()), group, jitter, timeout,
        timeout_disabled, args, kwargs, error_handler (by identity), mode, and
        predicate (by equality — lambdas/closures compare by identity).
        Does not compare runtime state (db_id, next_run, sort_index, _scheduler,
        _dequeued, owner, or any other mutable runtime field).

        Two jobs with identical callable/trigger/args but different groups are distinct
        logical jobs and will not match.
        """
        if self.trigger is not None and other.trigger is not None:
            triggers_match = self.trigger.trigger_id() == other.trigger.trigger_id()
        else:
            triggers_match = self.trigger is other.trigger
        return (
            self.job == other.job
            and triggers_match
            and self.group == other.group
            and self.jitter == other.jitter
            and self.timeout == other.timeout
            and self.timeout_disabled == other.timeout_disabled
            and self.args == other.args
            and self.kwargs == other.kwargs
            and self.error_handler is other.error_handler
            and self.mode is other.mode
            and self.predicate == other.predicate
        )

    def diff_fields(self, other: "ScheduledJob") -> list[str]:
        """Return a list of configuration field names that differ between two jobs.

        Compares the same fields as ``matches()`` — callable, trigger, group,
        jitter, timeout, timeout_disabled, args, kwargs, error_handler, mode,
        and predicate.
        """
        changed: list[str] = []
        if self.job != other.job:
            changed.append("job")
        self_tid = self.trigger.trigger_id() if self.trigger is not None else None
        other_tid = other.trigger.trigger_id() if other.trigger is not None else None
        if self_tid != other_tid:
            changed.append("trigger")
        if self.group != other.group:
            changed.append("group")
        if self.jitter != other.jitter:
            changed.append("jitter")
        if self.timeout != other.timeout:
            changed.append("timeout")
        if self.timeout_disabled != other.timeout_disabled:
            changed.append("timeout_disabled")
        if self.args != other.args:
            changed.append("args")
        if self.kwargs != other.kwargs:
            changed.append("kwargs")
        if self.error_handler is not other.error_handler:
            changed.append("error_handler")
        if self.mode is not other.mode:
            changed.append("mode")
        if self.predicate != other.predicate:
            changed.append("predicate")
        return changed

    def set_app_error_handler_resolver(self, resolver: "Callable[[], SchedulerErrorHandlerType | None]") -> None:
        """Set the closure that resolves the app-level error handler at dispatch time."""
        self.app_error_handler_resolver = resolver

    def cancel(self) -> None:
        """Cancel the job by delegating to the owning Scheduler.

        Raises:
            RuntimeError: When called on a job that has not been registered with a Scheduler.
                Use ``Scheduler.cancel_job(job)`` directly, or register the job first.
        """
        if self._scheduler is None:
            raise RuntimeError(
                "cancel() called on a job not registered with a Scheduler. "
                "Use Scheduler.cancel_job(job) or register the job first."
            )
        self._scheduler.cancel_job(self)

    def set_next_run(self, next_run: ZonedDateTime) -> None:
        """Update the next run timestamp, fire_at, and ordering metadata.

        Both ``next_run`` and ``fire_at`` are set to the rounded value. Call
        ``SchedulerService.apply_jitter_to_heap()`` after this to set a jittered
        ``fire_at`` when the job has ``jitter`` configured.
        """
        rounded = next_run.round("second")
        self.next_run = rounded
        self.fire_at = rounded
        self.sort_index = (rounded.timestamp_nanos(), id(self))

sort_index: tuple[int, int] = field(init=False, repr=False) class-attribute instance-attribute

Tuple of (next_run timestamp with nanoseconds, object id) for ordering in a priority queue.

owner_id: str = field(compare=False) class-attribute instance-attribute

Unique string identifier for the owner of the job, e.g., a component or integration name.

next_run: ZonedDateTime = field(compare=False) class-attribute instance-attribute

Unjittered logical fire time — used as previous_run in subsequent trigger calls.

fire_at: ZonedDateTime = field(init=False, compare=False) class-attribute instance-attribute

Actual dispatch time, including any jitter offset.

Equals next_run when no jitter is configured. Set by SchedulerService.apply_jitter_to_heap() at enqueue time when jitter > 0. The pop loop in _ScheduledJobQueue.pop_due_and_peek_next compares against fire_at (not next_run) to decide when to dispatch.

job: JobCallable = field(compare=False) class-attribute instance-attribute

The callable to execute when the job runs.

app_key: str = field(default='', compare=False) class-attribute instance-attribute

Configuration-level app key for DB registration (e.g., 'my_app'). Empty for non-App owners.

instance_index: int = field(default=0, compare=False) class-attribute instance-attribute

App instance index for DB registration. 0 for non-App owners.

instance_name: str | None = field(default=None, compare=False) class-attribute instance-attribute

App instance name, precomputed at registration time from the owning app's app_config.instance_name. None for framework-tier jobs and non-App owners.

trigger: TriggerProtocol | None = field(compare=False, default=None) class-attribute instance-attribute

The trigger that determines the job's schedule.

group: str | None = field(default=None, compare=False) class-attribute instance-attribute

Optional group name for grouping related jobs. Included in deduplication comparison.

jitter: float | None = field(default=None, compare=False) class-attribute instance-attribute

Seconds of random offset applied at enqueue time by SchedulerService.apply_jitter_to_heap().

Does not affect next_run (unjittered logical fire time). See the fire_at field on ScheduledJob for the actual dispatch time after jitter is applied.

timeout: float | None = field(default=None, compare=False) class-attribute instance-attribute

Per-job timeout in seconds. None means use the global default (config.scheduler_job_timeout_seconds). A positive float overrides the default. Validated at construction: must be positive when set.

timeout_disabled: bool = field(default=False, compare=False) class-attribute instance-attribute

When True, timeout enforcement is disabled for this job regardless of the global default.

name: str = field(default='', compare=False) class-attribute instance-attribute

Name for the job, used for identification. Required by all public Scheduler entry points (Scheduler.add_job() and Scheduler.schedule() raise SchedulerNameRequiredError if empty).

args: tuple[Any, ...] = field(default_factory=tuple, compare=False) class-attribute instance-attribute

Positional arguments to pass to the job callable.

kwargs: dict[str, Any] = field(default_factory=dict, compare=False) class-attribute instance-attribute

Keyword arguments to pass to the job callable.

error_handler: SchedulerErrorHandlerType | None = field(default=None, compare=False) class-attribute instance-attribute

Optional error handler for this job.

When set, this handler is invoked if the job raises an exception (including TimeoutError, but excluding CancelledError). Stored as-is for identity comparison in matches(). compare=False prevents Callable | None from corrupting the @dataclass(order=True) heap ordering.

db_id: int | None = field(default=None, compare=False) class-attribute instance-attribute

Database row ID for this job. Set by the executor after persistence; None until then.

source_location: str = field(default='', compare=False) class-attribute instance-attribute

Captured source location (file:line) of the user code that scheduled this job.

registration_source: str = field(default='', compare=False) class-attribute instance-attribute

Captured source code snippet of the scheduling call.

source_tier: SourceTier = field(default='app', compare=False) class-attribute instance-attribute

Whether this job originates from a user app or the framework itself.

mode: ExecutionMode = field(default=(ExecutionMode.SINGLE), compare=False) class-attribute instance-attribute

Resolved overlap mode for this job. Determines behavior when a prior invocation is still running as the next occurrence becomes due. Defaults to ExecutionMode.SINGLE so existing direct ScheduledJob(...) constructions in tests remain valid; the real resolution happens in Scheduler.schedule().

predicate: SchedulerPredicate | None = field(default=None, compare=False) class-attribute instance-attribute

Normalized predicate callable, or None when no where= was given.

Set once at registration time by Scheduler.schedule() — a single callable is stored directly, and a sequence of predicates is collapsed into a combinator. compare=False prevents Callable | None from corrupting the @dataclass(order=True) heap ordering.

predicate_invoker: CallableInvoker | None = field(default=None, repr=False, compare=False) class-attribute instance-attribute

Pre-built DI invoker for the predicate, or None when no where= was given.

Built once at registration time by Scheduler.schedule() using the shared DI layer (hassette.di) and passed alongside predicate. At dispatch time, predicate_invoker.invoke({ScheduledJob: job}) produces kwargs and the predicate is called with predicate(**kwargs).

guard: ExecutionModeGuard = field(init=False, compare=False) class-attribute instance-attribute

Per-job overlap state machine. Created in __post_init__ from mode. Lives for the full lifetime of the job, spanning all re-fires. compare=False prevents the guard from affecting heap ordering. init=False — do not pass directly; set via mode.

app_error_handler_resolver: Callable[[], SchedulerErrorHandlerType | None] | None = field(default=None, init=False, repr=False) class-attribute instance-attribute

Closure that resolves the app-level error handler at dispatch time.

pending_done: set[asyncio.Future[None]] = field(default_factory=set, init=False, repr=False, compare=False) class-attribute instance-attribute

Unresolved per-invocation completion futures for non-parallel modes.

Each run_job_with_guard call for single/restart/queued parks the outer dispatch task on a future that resolves when the invocation actually runs (or is dropped/released). A queued invocation accepted into the guard deque has no live child until drain time, so its future would hang forever if the job is cancelled first. dequeue_job, _remove_job, and _remove_jobs_by_owner call drain_pending_done(job.pending_done) after guard.release() so those dispatch tasks unwind.

mark_registered(db_id: int) -> None

Set the database ID. Called by SchedulerService.add_job() after persistence.

First call wins — a second call keeps the original id and logs a WARNING so a retry or double-registration is surfaced rather than silently swallowed. Mirrors Listener.mark_registered.

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

def mark_registered(self, db_id: int) -> None:
    """Set the database ID. Called by SchedulerService.add_job() after persistence.

    First call wins — a second call keeps the original id and logs a WARNING so a
    retry or double-registration is surfaced rather than silently swallowed.
    Mirrors ``Listener.mark_registered``.
    """
    if self.db_id is not None:
        LOGGER.warning(
            "ScheduledJob %s already registered with db_id=%s, ignoring new db_id=%s",
            self.name,
            self.db_id,
            db_id,
        )
        return
    self.db_id = db_id

matches(other: ScheduledJob) -> bool

Check whether two jobs represent the same logical configuration.

Compares callable, trigger (by trigger_id()), group, jitter, timeout, timeout_disabled, args, kwargs, error_handler (by identity), mode, and predicate (by equality — lambdas/closures compare by identity). Does not compare runtime state (db_id, next_run, sort_index, _scheduler, _dequeued, owner, or any other mutable runtime field).

Two jobs with identical callable/trigger/args but different groups are distinct logical jobs and will not match.

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

def matches(self, other: "ScheduledJob") -> bool:
    """Check whether two jobs represent the same logical configuration.

    Compares callable, trigger (by trigger_id()), group, jitter, timeout,
    timeout_disabled, args, kwargs, error_handler (by identity), mode, and
    predicate (by equality — lambdas/closures compare by identity).
    Does not compare runtime state (db_id, next_run, sort_index, _scheduler,
    _dequeued, owner, or any other mutable runtime field).

    Two jobs with identical callable/trigger/args but different groups are distinct
    logical jobs and will not match.
    """
    if self.trigger is not None and other.trigger is not None:
        triggers_match = self.trigger.trigger_id() == other.trigger.trigger_id()
    else:
        triggers_match = self.trigger is other.trigger
    return (
        self.job == other.job
        and triggers_match
        and self.group == other.group
        and self.jitter == other.jitter
        and self.timeout == other.timeout
        and self.timeout_disabled == other.timeout_disabled
        and self.args == other.args
        and self.kwargs == other.kwargs
        and self.error_handler is other.error_handler
        and self.mode is other.mode
        and self.predicate == other.predicate
    )

diff_fields(other: ScheduledJob) -> list[str]

Return a list of configuration field names that differ between two jobs.

Compares the same fields as matches() — callable, trigger, group, jitter, timeout, timeout_disabled, args, kwargs, error_handler, mode, and predicate.

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

def diff_fields(self, other: "ScheduledJob") -> list[str]:
    """Return a list of configuration field names that differ between two jobs.

    Compares the same fields as ``matches()`` — callable, trigger, group,
    jitter, timeout, timeout_disabled, args, kwargs, error_handler, mode,
    and predicate.
    """
    changed: list[str] = []
    if self.job != other.job:
        changed.append("job")
    self_tid = self.trigger.trigger_id() if self.trigger is not None else None
    other_tid = other.trigger.trigger_id() if other.trigger is not None else None
    if self_tid != other_tid:
        changed.append("trigger")
    if self.group != other.group:
        changed.append("group")
    if self.jitter != other.jitter:
        changed.append("jitter")
    if self.timeout != other.timeout:
        changed.append("timeout")
    if self.timeout_disabled != other.timeout_disabled:
        changed.append("timeout_disabled")
    if self.args != other.args:
        changed.append("args")
    if self.kwargs != other.kwargs:
        changed.append("kwargs")
    if self.error_handler is not other.error_handler:
        changed.append("error_handler")
    if self.mode is not other.mode:
        changed.append("mode")
    if self.predicate != other.predicate:
        changed.append("predicate")
    return changed

set_app_error_handler_resolver(resolver: Callable[[], SchedulerErrorHandlerType | None]) -> None

Set the closure that resolves the app-level error handler at dispatch time.

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

def set_app_error_handler_resolver(self, resolver: "Callable[[], SchedulerErrorHandlerType | None]") -> None:
    """Set the closure that resolves the app-level error handler at dispatch time."""
    self.app_error_handler_resolver = resolver

cancel() -> None

Cancel the job by delegating to the owning Scheduler.

Raises:

Type	Description
RuntimeError	When called on a job that has not been registered with a Scheduler. Use Scheduler.cancel_job(job) directly, or register the job first.

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

def cancel(self) -> None:
    """Cancel the job by delegating to the owning Scheduler.

    Raises:
        RuntimeError: When called on a job that has not been registered with a Scheduler.
            Use ``Scheduler.cancel_job(job)`` directly, or register the job first.
    """
    if self._scheduler is None:
        raise RuntimeError(
            "cancel() called on a job not registered with a Scheduler. "
            "Use Scheduler.cancel_job(job) or register the job first."
        )
    self._scheduler.cancel_job(self)

set_next_run(next_run: ZonedDateTime) -> None

Update the next run timestamp, fire_at, and ordering metadata.

Both next_run and fire_at are set to the rounded value. Call SchedulerService.apply_jitter_to_heap() after this to set a jittered fire_at when the job has jitter configured.

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

def set_next_run(self, next_run: ZonedDateTime) -> None:
    """Update the next run timestamp, fire_at, and ordering metadata.

    Both ``next_run`` and ``fire_at`` are set to the rounded value. Call
    ``SchedulerService.apply_jitter_to_heap()`` after this to set a jittered
    ``fire_at`` when the job has ``jitter`` configured.
    """
    rounded = next_run.round("second")
    self.next_run = rounded
    self.fire_at = rounded
    self.sort_index = (rounded.timestamp_nanos(), id(self))

SchedulerErrorContext dataclass

Bases: ErrorContext

Context passed to scheduler error handlers when a job raises an exception.

Attributes:

Name	Type	Description
exception	BaseException	The exception that was raised by the job.
traceback	str	Formatted traceback string.
job_name	str	The name of the job function that raised the exception.
job_group	str | None	The group the job belongs to, or None if ungrouped.
args	tuple[Any, ...]	Positional arguments the job was scheduled with.
kwargs	dict[str, Any]	Keyword arguments the job was scheduled with.

Source code in src/hassette/scheduler/error_context.py

@dataclass(frozen=True)
class SchedulerErrorContext(ErrorContext):
    """Context passed to scheduler error handlers when a job raises an exception.

    Attributes:
        exception: The exception that was raised by the job.
        traceback: Formatted traceback string.
        job_name: The name of the job function that raised the exception.
        job_group: The group the job belongs to, or None if ungrouped.
        args: Positional arguments the job was scheduled with.
        kwargs: Keyword arguments the job was scheduled with.
    """

    job_name: str
    job_group: str | None
    args: tuple[Any, ...]
    kwargs: dict[str, Any]

    @property
    def domain_label(self) -> str:
        return f"job={self.job_name}"

Scheduler

Bases: Resource

Scheduler resource for managing scheduled jobs.

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

class Scheduler(Resource):
    """Scheduler resource for managing scheduled jobs."""

    scheduler_service: SchedulerServiceProtocol
    """The scheduler service instance."""

    sync: SchedulerSyncFacade
    """Synchronous facade for scheduling jobs from sync code (e.g. ``AppSync`` hooks)."""

    _jobs_by_name: dict[str, "ScheduledJob"]
    """Tracks jobs by name for uniqueness validation within this scheduler instance."""

    _jobs_by_group: dict[str, set["ScheduledJob"]]
    """Tracks jobs by group for bulk cancellation.

    Uses ``set`` for O(1) membership test and discard. ``ScheduledJob.__hash__``
    is based on ``job_id``, which is unique and immutable after construction.
    """

    def __init__(self, hassette: "Hassette", *, parent: Resource | None = None) -> None:
        super().__init__(hassette, parent=parent)
        assert self.parent is not None, (
            "Scheduler requires a parent Resource for telemetry identity (app_key/source_tier)"
        )
        self.scheduler_service = self.hassette.scheduler_service
        self._jobs_by_name = {}
        self._jobs_by_group: dict[str, set[ScheduledJob]] = {}
        self._error_handler: SchedulerErrorHandlerType | None = None
        self.sync = self.add_child(SchedulerSyncFacade, scheduler=self)

        # Register removal callback so exhausted one-shot jobs are removed from _jobs_by_group
        # automatically when SchedulerService removes them after firing.
        self.scheduler_service.register_removal_callback(self.owner_id, self._on_job_removed)

    def _on_job_removed(self, job: "ScheduledJob") -> None:
        """Callback invoked by SchedulerService when a job is auto-exhausted.

        Keeps _jobs_by_group and _jobs_by_name in sync when SchedulerService removes a
        one-shot job after it fires or when a job is dequeued via cancel_job.
        """
        self._jobs_by_name.pop(job.name, None)
        if job.group is not None:
            group_set = self._jobs_by_group.get(job.group)
            if group_set is not None:
                group_set.discard(job)
                if not group_set:
                    del self._jobs_by_group[job.group]

    async def on_initialize(self) -> None:
        self._error_handler = None
        mark_ready(self, reason="Scheduler initialized")

    async def on_shutdown(self) -> None:
        await self._remove_all_jobs()
        self.scheduler_service.deregister_removal_callback(self.owner_id)

    def on_error(self, handler: "SchedulerErrorHandlerType") -> None:
        """Register an app-level error handler for this scheduler.

        The handler is called when any job on this scheduler raises an exception
        (including ``TimeoutError``) and the job does not have its own
        per-registration error handler.

        This is an app-level fallback — it is resolved at dispatch time, not at job
        registration time. A later call to ``on_error()`` replaces any previously
        registered handler.

        Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near
        app shutdown may be cancelled before they complete. Do not rely on error handlers
        for delivery-critical alerting during system teardown.

        Args:
            handler: A sync or async callable that accepts a
                :class:`~hassette.scheduler.error_context.SchedulerErrorContext`.
        """
        self._error_handler = handler

    @property
    def config_log_level(self) -> LOG_LEVEL_TYPE:
        """Return the log level from the config for this resource."""
        return self.hassette.config.logging.scheduler_service

    def add_job(
        self, job: "ScheduledJob", *, if_exists: IfExistsPolicy = "error"
    ) -> "Coroutine[Any, Any, ScheduledJob]":
        """Add a job to the scheduler.

        Must be awaited. Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        DB registration is awaited inline — ``job.db_id`` is set before this
        method returns, eliminating the window where a job fires with
        ``db_id=None``.

        Args:
            job: The job to add.
            if_exists: Behavior when a job with the same name already exists.
                ``"error"`` (default) raises ``ValueError``.
                ``"skip"`` returns the existing job if it matches; raises
                ``ValueError`` if the name matches but the configuration differs.
                ``"replace"`` cancels the existing job (recording it as cancelled
                in telemetry) and registers the new job in its place.

        Returns:
            The added job, or the existing job when ``if_exists="skip"`` and a
            matching job is already registered. ``job.db_id`` is a valid
            integer immediately on return.

        Raises:
            TypeError: If job is not a ScheduledJob.
            SchedulerNameRequiredError: If ``job.name`` is empty.
            ValueError: If a job with the same name already exists and either
                ``if_exists="error"`` or the existing job's configuration differs.
        """
        # Synchronous validation runs before the handle is constructed (design Edge Cases).
        if not isinstance(job, ScheduledJob):
            raise TypeError(f"Expected ScheduledJob, got {type(job).__name__}")
        if not job.name:
            raise SchedulerNameRequiredError(callable_name(job.job), str(job.trigger))
        # Eager capture in the public def — user frame is live here (not inside the async body).
        # Returns a 2-tuple — unpack it. Two destinations: guard_await (warning attribution) AND
        # _add_job (backfills job.source_location / registration_source for telemetry when empty).
        source_location, registration_source = capture_registration_source()
        # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
        return guard_await(
            self._add_job(
                job,
                if_exists=if_exists,
                source_location=source_location,
                registration_source=registration_source or "",
            ),
            owner=self.parent,
            source_location=source_location,
            method_name="add_job",
        )

    async def _add_job(
        self,
        job: "ScheduledJob",
        *,
        if_exists: IfExistsPolicy = "error",
        source_location: str = "",
        registration_source: str = "",
    ) -> "ScheduledJob":
        """Async body for add_job: duplicate-name check + registry mutations + DB registration.

        Duplicate-name handling and registry mutations must not run for a never-awaited
        call — they would pollute _jobs_by_name/_jobs_by_group. Matches the bus
        _resolve_and_register split where collision check lives in the awaited coroutine.

        source_location and registration_source are captured in the public def (user
        frame is live there) and threaded here for telemetry backfill.
        """
        # Empty string is the "not set" sentinel (ScheduledJob fields default to "").
        if not job.source_location:
            job.source_location = source_location
            job.registration_source = registration_source
        existing = self._jobs_by_name.get(job.name)
        if existing is not None:
            if if_exists == "replace":
                self.logger.debug("Replacing existing job '%s' (cancelling old, registering new)", job.name)
                self.cancel_job(existing)
            elif if_exists == "skip" and existing.matches(job):
                return existing
            elif if_exists == "skip":
                changed_fields = existing.diff_fields(job)
                raise ValueError(
                    f"A job named '{job.name}' already exists but its configuration has changed "
                    f"(changed fields: {', '.join(changed_fields)})"
                )
            else:
                raise ValueError(
                    f"A job named '{job.name}' already exists in scheduler for '{self.owner_id}'. "
                    "Job names must be unique per scheduler instance."
                )

        self._jobs_by_name[job.name] = job
        job._scheduler = self

        if job.group is not None:
            if job.group not in self._jobs_by_group:
                self._jobs_by_group[job.group] = set()
            self._jobs_by_group[job.group].add(job)

        job.set_app_error_handler_resolver(lambda: self._error_handler)
        await self.scheduler_service.add_job(job)

        return job

    def cancel_job(self, job: "ScheduledJob") -> None:
        """Cancel an individual job and persist the cancellation to the database.

        Idempotent: a second cancel on the same job is a silent no-op. Raises
        ``ValueError`` if the job belongs to a different scheduler instance.
        Spawns a durable ``mark_job_cancelled`` DB write (when ``db_id`` is set),
        dequeues the job from the service, and sets ``job._dequeued = True``.

        Must NOT call ``job.cancel()`` internally — that delegates back here and
        would cause infinite recursion.

        Args:
            job: The job to cancel.

        Raises:
            ValueError: If the job belongs to a different scheduler instance.
        """
        if job._dequeued:
            return  # idempotent — already cancelled
        if job._scheduler is not self:
            raise ValueError(
                f"cancel_job() called with a job belonging to a different scheduler "
                f"(job owner: {job._scheduler}, this scheduler: {self})"
            )
        if job.db_id is not None:
            # Spawn on scheduler_service.task_bucket (not self.task_bucket) so the
            # DB write survives Scheduler resource shutdown — the service's lifecycle
            # extends past the resource's cleanup phase.
            self.scheduler_service.task_bucket.spawn(
                self.scheduler_service.mark_job_cancelled(job.db_id),
                name="scheduler:mark_job_cancelled",
            )
        self.scheduler_service.dequeue_job(job)

    def _remove_all_jobs(self) -> asyncio.Task:
        """Remove all jobs for the owner of this scheduler."""
        self._jobs_by_name.clear()
        self._jobs_by_group.clear()
        return self.scheduler_service.remove_jobs_by_owner(self.owner_id)

    def cancel_group(self, group: str) -> None:
        """Cancel all jobs in the given group.

        Delegates to ``cancel_job`` per-member, which handles the DB write,
        dequeue, and ``_dequeued`` flag. Dict cleanup (``_jobs_by_group`` and
        ``_jobs_by_name``) is handled by the ``_on_job_removed`` callback
        fired by ``scheduler_service.dequeue_job``. No-op if the group does
        not exist.

        Args:
            group: The group name to cancel.
        """
        jobs = list(self._jobs_by_group.get(group, set()))
        for job in jobs:
            self.cancel_job(job)

    def list_jobs(self, group: str | None = None) -> list["ScheduledJob"]:
        """Return all or group-filtered jobs.

        Args:
            group: If provided, return only jobs in this group.
                If ``None`` (default), return all jobs.

        Returns:
            List of ScheduledJob instances.
        """
        if group is None:
            return list(self._jobs_by_name.values())
        return list(self._jobs_by_group.get(group, set()))

    def get_job_db_ids(self) -> list[int]:
        """Return the DB IDs of all registered jobs.

        Used by post-ready reconciliation in ``AppLifecycleService.initialize_instances()``
        to build the ``live_job_ids`` set. With synchronous registration, all jobs
        have a ``db_id`` set by the time ``on_initialize`` completes. The
        ``db_id is not None`` guard is kept as a defensive filter.

        Returns:
            List of integer DB row IDs for registered jobs.
        """
        return [job.db_id for job in self._jobs_by_name.values() if job.db_id is not None]

    def schedule(
        self,
        func: "JobCallable",
        trigger: "TriggerProtocol",
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> "Coroutine[Any, Any, ScheduledJob]":
        """Schedule a job using a trigger object.

        Must be awaited. Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        This is the primary entry point for scheduling. All convenience methods
        (``run_in``, ``run_every``, ``run_daily``, etc.) delegate here.

        Args:
            func: The function to run.
            trigger: A trigger object implementing ``TriggerProtocol``. Determines
                both the first run time and subsequent recurrences.
            name: Required stable name for the job. Used for uniqueness validation
                within this scheduler instance and for logging/telemetry.
            group: Optional group name for bulk management (see ``cancel_group``).
            jitter: Optional seconds of random offset to apply at enqueue time.
                Jitter is applied via ``SchedulerService.apply_jitter_to_heap`` on enqueue.
                See the ``fire_at`` field on ``ScheduledJob``.
            timeout: Per-job timeout in seconds. ``None`` uses the global default.
                A positive ``float`` overrides the default.
            timeout_disabled: When ``True``, timeout enforcement is disabled for this
                job regardless of the global default.
            mode: Overlap behavior when a prior invocation is still running as the next
                occurrence becomes due. One of ``"single"`` (skip the re-fire), ``"restart"``
                (cancel the running invocation and start fresh), ``"queued"`` (serialize
                re-fires in arrival order, up to the cap), or ``"parallel"`` (run concurrently).
                A raw string is coerced to ``ExecutionMode``; an invalid string raises
                ``ValueError`` naming the valid values. Omitted (``None``) resolves
                tier-aware: ``"single"`` for app-tier jobs, ``"parallel"`` for framework-tier
                jobs. Accepted on one-shot schedules (``run_in``/``run_once``) for API
                uniformity, but has no overlap effect since one-shots never re-fire.
            on_error: Optional per-job error handler. When set, this handler is
                invoked if the job raises an exception (including ``TimeoutError``,
                but excluding ``CancelledError``). Overrides the app-level handler
                set via ``on_error()``.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) evaluated at dispatch
                time, before the handler runs. Predicate signatures are inspected via
                the shared DI layer (``hassette.di``): a parameter annotated as
                ``ScheduledJob`` receives the job instance at dispatch time; unannotated
                predicates are called with zero arguments. A sequence is collapsed into
                a single combinator that ANDs all members — each member keeps the
                single-predicate contract. Predicates must be synchronous; async
                callables raise ``TypeError``. When the predicate returns ``False``,
                the handler does not run and a ``'skipped'`` execution is recorded.
                When the predicate raises, the handler does not run, an ``'error'``
                execution is recorded, and the job's ``on_error`` handler is invoked.

        Returns:
            The scheduled job. ``job.db_id`` is a valid integer immediately on return.

        Raises:
            TypeError: If ``trigger`` does not implement ``TriggerProtocol``, or if
                ``where`` is (or contains) an async callable.
            SchedulerNameRequiredError: If ``name`` is empty.
            DependencyInjectionError: If a predicate's signature is incompatible with
                DI (e.g. ``*args`` or positional-only parameters).
        """
        if not name:
            raise SchedulerNameRequiredError(callable_name(func), str(trigger))

        if jitter is not None and jitter < 0:
            raise ValueError("jitter must be non-negative")

        if not isinstance(trigger, TriggerProtocol):
            raise TypeError(
                f"trigger must implement TriggerProtocol; got {type(trigger).__name__}. "
                "Use hassette.scheduler.triggers (After, Once, Every, Daily, Cron)"
            )

        predicate, predicate_invoker = _normalize_where(where)

        parent = self.parent
        assert parent is not None
        app_key = parent.app_key
        instance_index = parent.index
        source_tier = parent.source_tier
        assert source_tier in ("app", "framework"), f"Invalid source_tier={source_tier!r} on {parent.class_name}"

        # Tier-aware default: an omitted mode (None) resolves to ``parallel`` for framework jobs
        # and ``single`` for app jobs. An explicit mode always wins. A raw string is coerced here
        # so an invalid value raises a clear ValueError at scheduling time.
        if mode is None:
            resolved_mode = ExecutionMode.PARALLEL if source_tier == "framework" else ExecutionMode.SINGLE
        elif isinstance(mode, ExecutionMode):
            resolved_mode = mode
        else:
            try:
                resolved_mode = ExecutionMode(mode)
            except ValueError as exc:
                valid = ", ".join(repr(m.value) for m in ExecutionMode)
                raise ValueError(f"Invalid execution mode {mode!r}; must be one of {valid}") from exc

        run_at = trigger.first_run_time(date_utils.now())

        # Resolve instance_name once at registration so the executor hot path reads it off the
        # command instead of traversing app_handler per execution.
        instance_name = parent.instance_name

        job = ScheduledJob(
            owner_id=self.owner_id,
            next_run=run_at,
            job=func,
            trigger=trigger,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            args=tuple(args) if args else (),
            kwargs=dict(kwargs) if kwargs else {},
            error_handler=on_error,
            app_key=app_key,
            instance_index=instance_index,
            instance_name=instance_name,
            source_tier=source_tier,
            mode=resolved_mode,
            predicate=predicate,
            predicate_invoker=predicate_invoker,
        )
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at add_job (the true primary). See design/071.
        # source_location/registration_source are NOT passed here; add_job backfills them.
        return self.add_job(job, if_exists=if_exists)

    def run_in(
        self,
        func: "JobCallable",
        delay: float,
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> "Coroutine[Any, Any, ScheduledJob]":
        """Schedule a job to run after a fixed delay (one-shot).

        Must be awaited. Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        Args:
            func: The function to run.
            delay: The delay in seconds before running the job.
            name: Required stable name for the job.
            group: Optional group name.
            jitter: Optional seconds of random offset to apply at enqueue time.
                See ``schedule()`` for details.
            timeout: Per-job timeout in seconds. See ``schedule()`` for details.
            timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
            mode: Overlap mode. Accepted for API uniformity; has no overlap effect for
                one-shot jobs since they never re-fire. See ``schedule()`` for the four
                values, tier-aware default, and string coercion rules.
            on_error: Optional per-job error handler. See ``schedule()`` for details.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) gating execution.
                See ``schedule()`` for details.

        Returns:
            The scheduled job.
        """
        trigger = After(seconds=float(delay))
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at add_job (the true primary). See design/071.
        return self.schedule(
            func,
            trigger,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )

    def run_once(
        self,
        func: "JobCallable",
        at: str | ZonedDateTime,
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        if_past: Literal["tomorrow", "error"] = "tomorrow",
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> "Coroutine[Any, Any, ScheduledJob]":
        """Schedule a job to run once at a specific wall-clock time (one-shot).

        Must be awaited. Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        Args:
            func: The function to run.
            at: Target time. A ``"HH:MM"`` string (today in the configured
                timezone — falls back to the process timezone when unset —
                or tomorrow if already past) or a ``ZonedDateTime``.
            name: Required stable name for the job.
            group: Optional group name.
            jitter: Optional seconds of random offset to apply at enqueue time.
                See ``schedule()`` for details.
            timeout: Per-job timeout in seconds. See ``schedule()`` for details.
            timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
            if_past: Behaviour when the target time is in the past at construction
                time. ``"tomorrow"`` (default) defers by one day. ``"error"`` raises
                ``ValueError``. For ``ZonedDateTime`` inputs, ``if_past`` has no
                effect — the job always fires immediately if the instant is in the past.
            mode: Overlap mode. Accepted for API uniformity; has no overlap effect for
                one-shot jobs since they never re-fire. See ``schedule()`` for the four
                values, tier-aware default, and string coercion rules.
            on_error: Optional per-job error handler. See ``schedule()`` for details.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) gating execution.
                See ``schedule()`` for details.

        Returns:
            The scheduled job.
        """
        trigger = Once(at=at, if_past=if_past)
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at add_job (the true primary). See design/071.
        return self.schedule(
            func,
            trigger,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )

    def run_every(
        self,
        func: "JobCallable",
        hours: float = 0,
        minutes: float = 0,
        seconds: float = 0,
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> "Coroutine[Any, Any, ScheduledJob]":
        """Schedule a job to run at a fixed interval.

        Must be awaited. Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        Args:
            func: The function to run.
            hours: Interval hours component.
            minutes: Interval minutes component.
            seconds: Interval seconds component.
            name: Required stable name for the job.
            group: Optional group name.
            jitter: Optional seconds of random offset to apply at enqueue time.
                See ``schedule()`` for details.
            timeout: Per-job timeout in seconds. See ``schedule()`` for details.
            timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
            mode: Overlap behavior when a prior invocation is still running as the next
                tick becomes due. See ``schedule()`` for the four values, tier-aware
                default, and string coercion rules.
            on_error: Optional per-job error handler. See ``schedule()`` for details.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) gating execution.
                See ``schedule()`` for details.

        Returns:
            The scheduled job.
        """
        trigger = Every(hours=hours, minutes=minutes, seconds=seconds)
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at add_job (the true primary). See design/071.
        return self.schedule(
            func,
            trigger,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )

    def run_minutely(
        self,
        func: "JobCallable",
        minutes: int = 1,
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> "Coroutine[Any, Any, ScheduledJob]":
        """Schedule a job to run every N minutes.

        Must be awaited. Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        Args:
            func: The function to run.
            minutes: The minute interval (must be >= 1).
            name: Required stable name for the job.
            group: Optional group name.
            jitter: Optional seconds of random offset to apply at enqueue time.
                See ``schedule()`` for details.
            timeout: Per-job timeout in seconds. See ``schedule()`` for details.
            timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
            mode: Overlap behavior when a prior invocation is still running as the next
                tick becomes due. See ``schedule()`` for the four values, tier-aware
                default, and string coercion rules.
            on_error: Optional per-job error handler. See ``schedule()`` for details.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) gating execution.
                See ``schedule()`` for details.

        Returns:
            The scheduled job.
        """
        if minutes < 1:
            raise ValueError("Minute interval must be at least 1")
        trigger = Every(minutes=minutes)
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at add_job (the true primary). See design/071.
        return self.schedule(
            func,
            trigger,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )

    def run_hourly(
        self,
        func: "JobCallable",
        hours: int = 1,
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> "Coroutine[Any, Any, ScheduledJob]":
        """Schedule a job to run every N hours.

        Must be awaited. Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        Args:
            func: The function to run.
            hours: The hour interval (must be >= 1).
            name: Required stable name for the job.
            group: Optional group name.
            jitter: Optional seconds of random offset to apply at enqueue time.
                See ``schedule()`` for details.
            timeout: Per-job timeout in seconds. See ``schedule()`` for details.
            timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
            mode: Overlap behavior when a prior invocation is still running as the next
                tick becomes due. See ``schedule()`` for the four values, tier-aware
                default, and string coercion rules.
            on_error: Optional per-job error handler. See ``schedule()`` for details.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) gating execution.
                See ``schedule()`` for details.

        Returns:
            The scheduled job.
        """
        if hours < 1:
            raise ValueError("Hour interval must be at least 1")
        trigger = Every(hours=hours)
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at add_job (the true primary). See design/071.
        return self.schedule(
            func,
            trigger,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )

    def run_daily(
        self,
        func: "JobCallable",
        at: str = "00:00",
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> "Coroutine[Any, Any, ScheduledJob]":
        """Schedule a job to run once per day at a fixed wall-clock time.

        Must be awaited. Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        Uses a cron-based trigger internally to ensure DST-correct, wall-clock-aligned
        scheduling. This avoids the 24-hour drift bug of interval-based daily scheduling.

        Args:
            func: The function to run.
            at: Target wall-clock time in ``"HH:MM"`` format (default ``"00:00"``).
            name: Required stable name for the job.
            group: Optional group name.
            jitter: Optional seconds of random offset to apply at enqueue time.
                See ``schedule()`` for details.
            timeout: Per-job timeout in seconds. See ``schedule()`` for details.
            timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
            mode: Overlap behavior when a prior invocation is still running as the next
                tick becomes due. See ``schedule()`` for the four values, tier-aware
                default, and string coercion rules.
            on_error: Optional per-job error handler. See ``schedule()`` for details.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) gating execution.
                See ``schedule()`` for details.

        Returns:
            The scheduled job.
        """
        trigger = Daily(at=at)
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at add_job (the true primary). See design/071.
        return self.schedule(
            func,
            trigger,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )

    def run_cron(
        self,
        func: "JobCallable",
        expression: str,
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> "Coroutine[Any, Any, ScheduledJob]":
        """Schedule a job using a cron expression.

        Must be awaited. Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        Accepts both 5-field (standard Unix cron: ``minute hour dom month dow``)
        and 6-field expressions (seconds appended as a 6th field per croniter
        convention: ``minute hour dom month dow second``).

        Args:
            func: The function to run.
            expression: A valid 5- or 6-field cron expression.
            name: Required stable name for the job.
            group: Optional group name.
            jitter: Optional seconds of random offset to apply at enqueue time.
                See ``schedule()`` for details.
            timeout: Per-job timeout in seconds. See ``schedule()`` for details.
            timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
            mode: Overlap behavior when a prior invocation is still running as the next
                tick becomes due. See ``schedule()`` for the four values, tier-aware
                default, and string coercion rules.
            on_error: Optional per-job error handler. See ``schedule()`` for details.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) gating execution.
                See ``schedule()`` for details.

        Returns:
            The scheduled job.

        Raises:
            ValueError: If the cron expression is syntactically invalid.
        """
        trigger = Cron(expression)
        # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
        # The single guard_await lives at add_job (the true primary). See design/071.
        return self.schedule(
            func,
            trigger,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )

scheduler_service: SchedulerServiceProtocol = self.hassette.scheduler_service instance-attribute

The scheduler service instance.

sync: SchedulerSyncFacade = self.add_child(SchedulerSyncFacade, scheduler=self) instance-attribute

Synchronous facade for scheduling jobs from sync code (e.g. AppSync hooks).

config_log_level: LOG_LEVEL_TYPE property

Return the log level from the config for this resource.

on_error(handler: SchedulerErrorHandlerType) -> None

Register an app-level error handler for this scheduler.

The handler is called when any job on this scheduler raises an exception (including TimeoutError) and the job does not have its own per-registration error handler.

This is an app-level fallback — it is resolved at dispatch time, not at job registration time. A later call to on_error() replaces any previously registered handler.

Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near app shutdown may be cancelled before they complete. Do not rely on error handlers for delivery-critical alerting during system teardown.

Parameters:

Name	Type	Description	Default
handler	SchedulerErrorHandlerType	A sync or async callable that accepts a :class:~hassette.scheduler.error_context.SchedulerErrorContext.	required

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

def on_error(self, handler: "SchedulerErrorHandlerType") -> None:
    """Register an app-level error handler for this scheduler.

    The handler is called when any job on this scheduler raises an exception
    (including ``TimeoutError``) and the job does not have its own
    per-registration error handler.

    This is an app-level fallback — it is resolved at dispatch time, not at job
    registration time. A later call to ``on_error()`` replaces any previously
    registered handler.

    Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near
    app shutdown may be cancelled before they complete. Do not rely on error handlers
    for delivery-critical alerting during system teardown.

    Args:
        handler: A sync or async callable that accepts a
            :class:`~hassette.scheduler.error_context.SchedulerErrorContext`.
    """
    self._error_handler = handler

add_job(job: ScheduledJob, *, if_exists: IfExistsPolicy = 'error') -> Coroutine[Any, Any, ScheduledJob]

Add a job to the scheduler.

Must be awaited. Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

DB registration is awaited inline — job.db_id is set before this method returns, eliminating the window where a job fires with db_id=None.

Parameters:

Name	Type	Description	Default
job	ScheduledJob	The job to add.	required
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. "error" (default) raises ValueError. "skip" returns the existing job if it matches; raises ValueError if the name matches but the configuration differs. "replace" cancels the existing job (recording it as cancelled in telemetry) and registers the new job in its place.	'error'

Returns:

Type	Description
Coroutine[Any, Any, ScheduledJob]	The added job, or the existing job when if_exists="skip" and a
Coroutine[Any, Any, ScheduledJob]	matching job is already registered. job.db_id is a valid
Coroutine[Any, Any, ScheduledJob]	integer immediately on return.

Raises:

Type	Description
TypeError	If job is not a ScheduledJob.
SchedulerNameRequiredError	If job.name is empty.
ValueError	If a job with the same name already exists and either if_exists="error" or the existing job's configuration differs.

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

def add_job(
    self, job: "ScheduledJob", *, if_exists: IfExistsPolicy = "error"
) -> "Coroutine[Any, Any, ScheduledJob]":
    """Add a job to the scheduler.

    Must be awaited. Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    DB registration is awaited inline — ``job.db_id`` is set before this
    method returns, eliminating the window where a job fires with
    ``db_id=None``.

    Args:
        job: The job to add.
        if_exists: Behavior when a job with the same name already exists.
            ``"error"`` (default) raises ``ValueError``.
            ``"skip"`` returns the existing job if it matches; raises
            ``ValueError`` if the name matches but the configuration differs.
            ``"replace"`` cancels the existing job (recording it as cancelled
            in telemetry) and registers the new job in its place.

    Returns:
        The added job, or the existing job when ``if_exists="skip"`` and a
        matching job is already registered. ``job.db_id`` is a valid
        integer immediately on return.

    Raises:
        TypeError: If job is not a ScheduledJob.
        SchedulerNameRequiredError: If ``job.name`` is empty.
        ValueError: If a job with the same name already exists and either
            ``if_exists="error"`` or the existing job's configuration differs.
    """
    # Synchronous validation runs before the handle is constructed (design Edge Cases).
    if not isinstance(job, ScheduledJob):
        raise TypeError(f"Expected ScheduledJob, got {type(job).__name__}")
    if not job.name:
        raise SchedulerNameRequiredError(callable_name(job.job), str(job.trigger))
    # Eager capture in the public def — user frame is live here (not inside the async body).
    # Returns a 2-tuple — unpack it. Two destinations: guard_await (warning attribution) AND
    # _add_job (backfills job.source_location / registration_source for telemetry when empty).
    source_location, registration_source = capture_registration_source()
    # Coroutine[...] supertype annotation is load-bearing — see hassette/utils/await_guard.py / design/071.
    return guard_await(
        self._add_job(
            job,
            if_exists=if_exists,
            source_location=source_location,
            registration_source=registration_source or "",
        ),
        owner=self.parent,
        source_location=source_location,
        method_name="add_job",
    )

cancel_job(job: ScheduledJob) -> None

Cancel an individual job and persist the cancellation to the database.

Idempotent: a second cancel on the same job is a silent no-op. Raises ValueError if the job belongs to a different scheduler instance. Spawns a durable mark_job_cancelled DB write (when db_id is set), dequeues the job from the service, and sets job._dequeued = True.

Must NOT call job.cancel() internally — that delegates back here and would cause infinite recursion.

Parameters:

Name	Type	Description	Default
job	ScheduledJob	The job to cancel.	required

Raises:

Type	Description
ValueError	If the job belongs to a different scheduler instance.

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

def cancel_job(self, job: "ScheduledJob") -> None:
    """Cancel an individual job and persist the cancellation to the database.

    Idempotent: a second cancel on the same job is a silent no-op. Raises
    ``ValueError`` if the job belongs to a different scheduler instance.
    Spawns a durable ``mark_job_cancelled`` DB write (when ``db_id`` is set),
    dequeues the job from the service, and sets ``job._dequeued = True``.

    Must NOT call ``job.cancel()`` internally — that delegates back here and
    would cause infinite recursion.

    Args:
        job: The job to cancel.

    Raises:
        ValueError: If the job belongs to a different scheduler instance.
    """
    if job._dequeued:
        return  # idempotent — already cancelled
    if job._scheduler is not self:
        raise ValueError(
            f"cancel_job() called with a job belonging to a different scheduler "
            f"(job owner: {job._scheduler}, this scheduler: {self})"
        )
    if job.db_id is not None:
        # Spawn on scheduler_service.task_bucket (not self.task_bucket) so the
        # DB write survives Scheduler resource shutdown — the service's lifecycle
        # extends past the resource's cleanup phase.
        self.scheduler_service.task_bucket.spawn(
            self.scheduler_service.mark_job_cancelled(job.db_id),
            name="scheduler:mark_job_cancelled",
        )
    self.scheduler_service.dequeue_job(job)

cancel_group(group: str) -> None

Cancel all jobs in the given group.

Delegates to cancel_job per-member, which handles the DB write, dequeue, and _dequeued flag. Dict cleanup (_jobs_by_group and _jobs_by_name) is handled by the _on_job_removed callback fired by scheduler_service.dequeue_job. No-op if the group does not exist.

Parameters:

Name	Type	Description	Default
group	str	The group name to cancel.	required

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

def cancel_group(self, group: str) -> None:
    """Cancel all jobs in the given group.

    Delegates to ``cancel_job`` per-member, which handles the DB write,
    dequeue, and ``_dequeued`` flag. Dict cleanup (``_jobs_by_group`` and
    ``_jobs_by_name``) is handled by the ``_on_job_removed`` callback
    fired by ``scheduler_service.dequeue_job``. No-op if the group does
    not exist.

    Args:
        group: The group name to cancel.
    """
    jobs = list(self._jobs_by_group.get(group, set()))
    for job in jobs:
        self.cancel_job(job)

list_jobs(group: str | None = None) -> list[ScheduledJob]

Return all or group-filtered jobs.

Parameters:

Name	Type	Description	Default
group	str | None	If provided, return only jobs in this group. If None (default), return all jobs.	None

Returns:

Type	Description
list[ScheduledJob]	List of ScheduledJob instances.

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

def list_jobs(self, group: str | None = None) -> list["ScheduledJob"]:
    """Return all or group-filtered jobs.

    Args:
        group: If provided, return only jobs in this group.
            If ``None`` (default), return all jobs.

    Returns:
        List of ScheduledJob instances.
    """
    if group is None:
        return list(self._jobs_by_name.values())
    return list(self._jobs_by_group.get(group, set()))

get_job_db_ids() -> list[int]

Return the DB IDs of all registered jobs.

Used by post-ready reconciliation in AppLifecycleService.initialize_instances() to build the live_job_ids set. With synchronous registration, all jobs have a db_id set by the time on_initialize completes. The db_id is not None guard is kept as a defensive filter.

Returns:

Type	Description
list[int]	List of integer DB row IDs for registered jobs.

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

def get_job_db_ids(self) -> list[int]:
    """Return the DB IDs of all registered jobs.

    Used by post-ready reconciliation in ``AppLifecycleService.initialize_instances()``
    to build the ``live_job_ids`` set. With synchronous registration, all jobs
    have a ``db_id`` set by the time ``on_initialize`` completes. The
    ``db_id is not None`` guard is kept as a defensive filter.

    Returns:
        List of integer DB row IDs for registered jobs.
    """
    return [job.db_id for job in self._jobs_by_name.values() if job.db_id is not None]

schedule(func: JobCallable, trigger: TriggerProtocol, *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> Coroutine[Any, Any, ScheduledJob]

Schedule a job using a trigger object.

Must be awaited. Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

This is the primary entry point for scheduling. All convenience methods (run_in, run_every, run_daily, etc.) delegate here.

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
trigger	TriggerProtocol	A trigger object implementing TriggerProtocol. Determines both the first run time and subsequent recurrences.	required
name	str	Required stable name for the job. Used for uniqueness validation within this scheduler instance and for logging/telemetry.	required
group	str | None	Optional group name for bulk management (see cancel_group).	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. Jitter is applied via SchedulerService.apply_jitter_to_heap on enqueue. See the fire_at field on ScheduledJob.	None
timeout	float | None	Per-job timeout in seconds. None uses the global default. A positive float overrides the default.	None
timeout_disabled	bool	When True, timeout enforcement is disabled for this job regardless of the global default.	False
mode	ExecutionMode | str | None	Overlap behavior when a prior invocation is still running as the next occurrence becomes due. One of "single" (skip the re-fire), "restart" (cancel the running invocation and start fresh), "queued" (serialize re-fires in arrival order, up to the cap), or "parallel" (run concurrently). A raw string is coerced to ExecutionMode; an invalid string raises ValueError naming the valid values. Omitted (None) resolves tier-aware: "single" for app-tier jobs, "parallel" for framework-tier jobs. Accepted on one-shot schedules (run_in/run_once) for API uniformity, but has no overlap effect since one-shots never re-fire.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. When set, this handler is invoked if the job raises an exception (including TimeoutError, but excluding CancelledError). Overrides the app-level handler set via on_error().	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) evaluated at dispatch time, before the handler runs. Predicate signatures are inspected via the shared DI layer (hassette.di): a parameter annotated as ScheduledJob receives the job instance at dispatch time; unannotated predicates are called with zero arguments. A sequence is collapsed into a single combinator that ANDs all members — each member keeps the single-predicate contract. Predicates must be synchronous; async callables raise TypeError. When the predicate returns False, the handler does not run and a 'skipped' execution is recorded. When the predicate raises, the handler does not run, an 'error' execution is recorded, and the job's on_error handler is invoked.	None

Returns:

Type	Description
Coroutine[Any, Any, ScheduledJob]	The scheduled job. job.db_id is a valid integer immediately on return.

Raises:

Type	Description
TypeError	If trigger does not implement TriggerProtocol, or if where is (or contains) an async callable.
SchedulerNameRequiredError	If name is empty.
DependencyInjectionError	If a predicate's signature is incompatible with DI (e.g. *args or positional-only parameters).

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

def schedule(
    self,
    func: "JobCallable",
    trigger: "TriggerProtocol",
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> "Coroutine[Any, Any, ScheduledJob]":
    """Schedule a job using a trigger object.

    Must be awaited. Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    This is the primary entry point for scheduling. All convenience methods
    (``run_in``, ``run_every``, ``run_daily``, etc.) delegate here.

    Args:
        func: The function to run.
        trigger: A trigger object implementing ``TriggerProtocol``. Determines
            both the first run time and subsequent recurrences.
        name: Required stable name for the job. Used for uniqueness validation
            within this scheduler instance and for logging/telemetry.
        group: Optional group name for bulk management (see ``cancel_group``).
        jitter: Optional seconds of random offset to apply at enqueue time.
            Jitter is applied via ``SchedulerService.apply_jitter_to_heap`` on enqueue.
            See the ``fire_at`` field on ``ScheduledJob``.
        timeout: Per-job timeout in seconds. ``None`` uses the global default.
            A positive ``float`` overrides the default.
        timeout_disabled: When ``True``, timeout enforcement is disabled for this
            job regardless of the global default.
        mode: Overlap behavior when a prior invocation is still running as the next
            occurrence becomes due. One of ``"single"`` (skip the re-fire), ``"restart"``
            (cancel the running invocation and start fresh), ``"queued"`` (serialize
            re-fires in arrival order, up to the cap), or ``"parallel"`` (run concurrently).
            A raw string is coerced to ``ExecutionMode``; an invalid string raises
            ``ValueError`` naming the valid values. Omitted (``None``) resolves
            tier-aware: ``"single"`` for app-tier jobs, ``"parallel"`` for framework-tier
            jobs. Accepted on one-shot schedules (``run_in``/``run_once``) for API
            uniformity, but has no overlap effect since one-shots never re-fire.
        on_error: Optional per-job error handler. When set, this handler is
            invoked if the job raises an exception (including ``TimeoutError``,
            but excluding ``CancelledError``). Overrides the app-level handler
            set via ``on_error()``.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) evaluated at dispatch
            time, before the handler runs. Predicate signatures are inspected via
            the shared DI layer (``hassette.di``): a parameter annotated as
            ``ScheduledJob`` receives the job instance at dispatch time; unannotated
            predicates are called with zero arguments. A sequence is collapsed into
            a single combinator that ANDs all members — each member keeps the
            single-predicate contract. Predicates must be synchronous; async
            callables raise ``TypeError``. When the predicate returns ``False``,
            the handler does not run and a ``'skipped'`` execution is recorded.
            When the predicate raises, the handler does not run, an ``'error'``
            execution is recorded, and the job's ``on_error`` handler is invoked.

    Returns:
        The scheduled job. ``job.db_id`` is a valid integer immediately on return.

    Raises:
        TypeError: If ``trigger`` does not implement ``TriggerProtocol``, or if
            ``where`` is (or contains) an async callable.
        SchedulerNameRequiredError: If ``name`` is empty.
        DependencyInjectionError: If a predicate's signature is incompatible with
            DI (e.g. ``*args`` or positional-only parameters).
    """
    if not name:
        raise SchedulerNameRequiredError(callable_name(func), str(trigger))

    if jitter is not None and jitter < 0:
        raise ValueError("jitter must be non-negative")

    if not isinstance(trigger, TriggerProtocol):
        raise TypeError(
            f"trigger must implement TriggerProtocol; got {type(trigger).__name__}. "
            "Use hassette.scheduler.triggers (After, Once, Every, Daily, Cron)"
        )

    predicate, predicate_invoker = _normalize_where(where)

    parent = self.parent
    assert parent is not None
    app_key = parent.app_key
    instance_index = parent.index
    source_tier = parent.source_tier
    assert source_tier in ("app", "framework"), f"Invalid source_tier={source_tier!r} on {parent.class_name}"

    # Tier-aware default: an omitted mode (None) resolves to ``parallel`` for framework jobs
    # and ``single`` for app jobs. An explicit mode always wins. A raw string is coerced here
    # so an invalid value raises a clear ValueError at scheduling time.
    if mode is None:
        resolved_mode = ExecutionMode.PARALLEL if source_tier == "framework" else ExecutionMode.SINGLE
    elif isinstance(mode, ExecutionMode):
        resolved_mode = mode
    else:
        try:
            resolved_mode = ExecutionMode(mode)
        except ValueError as exc:
            valid = ", ".join(repr(m.value) for m in ExecutionMode)
            raise ValueError(f"Invalid execution mode {mode!r}; must be one of {valid}") from exc

    run_at = trigger.first_run_time(date_utils.now())

    # Resolve instance_name once at registration so the executor hot path reads it off the
    # command instead of traversing app_handler per execution.
    instance_name = parent.instance_name

    job = ScheduledJob(
        owner_id=self.owner_id,
        next_run=run_at,
        job=func,
        trigger=trigger,
        name=name,
        group=group,
        jitter=jitter,
        timeout=timeout,
        timeout_disabled=timeout_disabled,
        args=tuple(args) if args else (),
        kwargs=dict(kwargs) if kwargs else {},
        error_handler=on_error,
        app_key=app_key,
        instance_index=instance_index,
        instance_name=instance_name,
        source_tier=source_tier,
        mode=resolved_mode,
        predicate=predicate,
        predicate_invoker=predicate_invoker,
    )
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at add_job (the true primary). See design/071.
    # source_location/registration_source are NOT passed here; add_job backfills them.
    return self.add_job(job, if_exists=if_exists)

run_in(func: JobCallable, delay: float, *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> Coroutine[Any, Any, ScheduledJob]

Schedule a job to run after a fixed delay (one-shot).

Must be awaited. Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
delay	float	The delay in seconds before running the job.	required
name	str	Required stable name for the job.	required
group	str | None	Optional group name.	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. See schedule() for details.	None
timeout	float | None	Per-job timeout in seconds. See schedule() for details.	None
timeout_disabled	bool	Disable timeout enforcement. See schedule() for details.	False
mode	ExecutionMode | str | None	Overlap mode. Accepted for API uniformity; has no overlap effect for one-shot jobs since they never re-fire. See schedule() for the four values, tier-aware default, and string coercion rules.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. See schedule() for details.	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) gating execution. See schedule() for details.	None

Returns:

Type	Description
Coroutine[Any, Any, ScheduledJob]	The scheduled job.

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

def run_in(
    self,
    func: "JobCallable",
    delay: float,
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> "Coroutine[Any, Any, ScheduledJob]":
    """Schedule a job to run after a fixed delay (one-shot).

    Must be awaited. Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    Args:
        func: The function to run.
        delay: The delay in seconds before running the job.
        name: Required stable name for the job.
        group: Optional group name.
        jitter: Optional seconds of random offset to apply at enqueue time.
            See ``schedule()`` for details.
        timeout: Per-job timeout in seconds. See ``schedule()`` for details.
        timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
        mode: Overlap mode. Accepted for API uniformity; has no overlap effect for
            one-shot jobs since they never re-fire. See ``schedule()`` for the four
            values, tier-aware default, and string coercion rules.
        on_error: Optional per-job error handler. See ``schedule()`` for details.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) gating execution.
            See ``schedule()`` for details.

    Returns:
        The scheduled job.
    """
    trigger = After(seconds=float(delay))
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at add_job (the true primary). See design/071.
    return self.schedule(
        func,
        trigger,
        name=name,
        group=group,
        jitter=jitter,
        timeout=timeout,
        timeout_disabled=timeout_disabled,
        mode=mode,
        on_error=on_error,
        if_exists=if_exists,
        args=args,
        kwargs=kwargs,
        where=where,
    )

run_once(func: JobCallable, at: str | ZonedDateTime, *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, if_past: Literal['tomorrow', 'error'] = 'tomorrow', mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> Coroutine[Any, Any, ScheduledJob]

Schedule a job to run once at a specific wall-clock time (one-shot).

Must be awaited. Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
at	str | ZonedDateTime	Target time. A "HH:MM" string (today in the configured timezone — falls back to the process timezone when unset — or tomorrow if already past) or a ZonedDateTime.	required
name	str	Required stable name for the job.	required
group	str | None	Optional group name.	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. See schedule() for details.	None
timeout	float | None	Per-job timeout in seconds. See schedule() for details.	None
timeout_disabled	bool	Disable timeout enforcement. See schedule() for details.	False
if_past	Literal['tomorrow', 'error']	Behaviour when the target time is in the past at construction time. "tomorrow" (default) defers by one day. "error" raises ValueError. For ZonedDateTime inputs, if_past has no effect — the job always fires immediately if the instant is in the past.	'tomorrow'
mode	ExecutionMode | str | None	Overlap mode. Accepted for API uniformity; has no overlap effect for one-shot jobs since they never re-fire. See schedule() for the four values, tier-aware default, and string coercion rules.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. See schedule() for details.	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) gating execution. See schedule() for details.	None

Returns:

Type	Description
Coroutine[Any, Any, ScheduledJob]	The scheduled job.

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

def run_once(
    self,
    func: "JobCallable",
    at: str | ZonedDateTime,
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    if_past: Literal["tomorrow", "error"] = "tomorrow",
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> "Coroutine[Any, Any, ScheduledJob]":
    """Schedule a job to run once at a specific wall-clock time (one-shot).

    Must be awaited. Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    Args:
        func: The function to run.
        at: Target time. A ``"HH:MM"`` string (today in the configured
            timezone — falls back to the process timezone when unset —
            or tomorrow if already past) or a ``ZonedDateTime``.
        name: Required stable name for the job.
        group: Optional group name.
        jitter: Optional seconds of random offset to apply at enqueue time.
            See ``schedule()`` for details.
        timeout: Per-job timeout in seconds. See ``schedule()`` for details.
        timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
        if_past: Behaviour when the target time is in the past at construction
            time. ``"tomorrow"`` (default) defers by one day. ``"error"`` raises
            ``ValueError``. For ``ZonedDateTime`` inputs, ``if_past`` has no
            effect — the job always fires immediately if the instant is in the past.
        mode: Overlap mode. Accepted for API uniformity; has no overlap effect for
            one-shot jobs since they never re-fire. See ``schedule()`` for the four
            values, tier-aware default, and string coercion rules.
        on_error: Optional per-job error handler. See ``schedule()`` for details.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) gating execution.
            See ``schedule()`` for details.

    Returns:
        The scheduled job.
    """
    trigger = Once(at=at, if_past=if_past)
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at add_job (the true primary). See design/071.
    return self.schedule(
        func,
        trigger,
        name=name,
        group=group,
        jitter=jitter,
        timeout=timeout,
        timeout_disabled=timeout_disabled,
        mode=mode,
        on_error=on_error,
        if_exists=if_exists,
        args=args,
        kwargs=kwargs,
        where=where,
    )

run_every(func: JobCallable, hours: float = 0, minutes: float = 0, seconds: float = 0, *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> Coroutine[Any, Any, ScheduledJob]

Schedule a job to run at a fixed interval.

Must be awaited. Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
hours	float	Interval hours component.	0
minutes	float	Interval minutes component.	0
seconds	float	Interval seconds component.	0
name	str	Required stable name for the job.	required
group	str | None	Optional group name.	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. See schedule() for details.	None
timeout	float | None	Per-job timeout in seconds. See schedule() for details.	None
timeout_disabled	bool	Disable timeout enforcement. See schedule() for details.	False
mode	ExecutionMode | str | None	Overlap behavior when a prior invocation is still running as the next tick becomes due. See schedule() for the four values, tier-aware default, and string coercion rules.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. See schedule() for details.	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) gating execution. See schedule() for details.	None

Returns:

Type	Description
Coroutine[Any, Any, ScheduledJob]	The scheduled job.

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

def run_every(
    self,
    func: "JobCallable",
    hours: float = 0,
    minutes: float = 0,
    seconds: float = 0,
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> "Coroutine[Any, Any, ScheduledJob]":
    """Schedule a job to run at a fixed interval.

    Must be awaited. Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    Args:
        func: The function to run.
        hours: Interval hours component.
        minutes: Interval minutes component.
        seconds: Interval seconds component.
        name: Required stable name for the job.
        group: Optional group name.
        jitter: Optional seconds of random offset to apply at enqueue time.
            See ``schedule()`` for details.
        timeout: Per-job timeout in seconds. See ``schedule()`` for details.
        timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
        mode: Overlap behavior when a prior invocation is still running as the next
            tick becomes due. See ``schedule()`` for the four values, tier-aware
            default, and string coercion rules.
        on_error: Optional per-job error handler. See ``schedule()`` for details.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) gating execution.
            See ``schedule()`` for details.

    Returns:
        The scheduled job.
    """
    trigger = Every(hours=hours, minutes=minutes, seconds=seconds)
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at add_job (the true primary). See design/071.
    return self.schedule(
        func,
        trigger,
        name=name,
        group=group,
        jitter=jitter,
        timeout=timeout,
        timeout_disabled=timeout_disabled,
        mode=mode,
        on_error=on_error,
        if_exists=if_exists,
        args=args,
        kwargs=kwargs,
        where=where,
    )

run_minutely(func: JobCallable, minutes: int = 1, *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> Coroutine[Any, Any, ScheduledJob]

Schedule a job to run every N minutes.

Must be awaited. Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
minutes	int	The minute interval (must be >= 1).	1
name	str	Required stable name for the job.	required
group	str | None	Optional group name.	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. See schedule() for details.	None
timeout	float | None	Per-job timeout in seconds. See schedule() for details.	None
timeout_disabled	bool	Disable timeout enforcement. See schedule() for details.	False
mode	ExecutionMode | str | None	Overlap behavior when a prior invocation is still running as the next tick becomes due. See schedule() for the four values, tier-aware default, and string coercion rules.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. See schedule() for details.	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) gating execution. See schedule() for details.	None

Returns:

Type	Description
Coroutine[Any, Any, ScheduledJob]	The scheduled job.

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

def run_minutely(
    self,
    func: "JobCallable",
    minutes: int = 1,
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> "Coroutine[Any, Any, ScheduledJob]":
    """Schedule a job to run every N minutes.

    Must be awaited. Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    Args:
        func: The function to run.
        minutes: The minute interval (must be >= 1).
        name: Required stable name for the job.
        group: Optional group name.
        jitter: Optional seconds of random offset to apply at enqueue time.
            See ``schedule()`` for details.
        timeout: Per-job timeout in seconds. See ``schedule()`` for details.
        timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
        mode: Overlap behavior when a prior invocation is still running as the next
            tick becomes due. See ``schedule()`` for the four values, tier-aware
            default, and string coercion rules.
        on_error: Optional per-job error handler. See ``schedule()`` for details.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) gating execution.
            See ``schedule()`` for details.

    Returns:
        The scheduled job.
    """
    if minutes < 1:
        raise ValueError("Minute interval must be at least 1")
    trigger = Every(minutes=minutes)
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at add_job (the true primary). See design/071.
    return self.schedule(
        func,
        trigger,
        name=name,
        group=group,
        jitter=jitter,
        timeout=timeout,
        timeout_disabled=timeout_disabled,
        mode=mode,
        on_error=on_error,
        if_exists=if_exists,
        args=args,
        kwargs=kwargs,
        where=where,
    )

run_hourly(func: JobCallable, hours: int = 1, *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> Coroutine[Any, Any, ScheduledJob]

Schedule a job to run every N hours.

Must be awaited. Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
hours	int	The hour interval (must be >= 1).	1
name	str	Required stable name for the job.	required
group	str | None	Optional group name.	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. See schedule() for details.	None
timeout	float | None	Per-job timeout in seconds. See schedule() for details.	None
timeout_disabled	bool	Disable timeout enforcement. See schedule() for details.	False
mode	ExecutionMode | str | None	Overlap behavior when a prior invocation is still running as the next tick becomes due. See schedule() for the four values, tier-aware default, and string coercion rules.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. See schedule() for details.	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) gating execution. See schedule() for details.	None

Returns:

Type	Description
Coroutine[Any, Any, ScheduledJob]	The scheduled job.

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

def run_hourly(
    self,
    func: "JobCallable",
    hours: int = 1,
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> "Coroutine[Any, Any, ScheduledJob]":
    """Schedule a job to run every N hours.

    Must be awaited. Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    Args:
        func: The function to run.
        hours: The hour interval (must be >= 1).
        name: Required stable name for the job.
        group: Optional group name.
        jitter: Optional seconds of random offset to apply at enqueue time.
            See ``schedule()`` for details.
        timeout: Per-job timeout in seconds. See ``schedule()`` for details.
        timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
        mode: Overlap behavior when a prior invocation is still running as the next
            tick becomes due. See ``schedule()`` for the four values, tier-aware
            default, and string coercion rules.
        on_error: Optional per-job error handler. See ``schedule()`` for details.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) gating execution.
            See ``schedule()`` for details.

    Returns:
        The scheduled job.
    """
    if hours < 1:
        raise ValueError("Hour interval must be at least 1")
    trigger = Every(hours=hours)
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at add_job (the true primary). See design/071.
    return self.schedule(
        func,
        trigger,
        name=name,
        group=group,
        jitter=jitter,
        timeout=timeout,
        timeout_disabled=timeout_disabled,
        mode=mode,
        on_error=on_error,
        if_exists=if_exists,
        args=args,
        kwargs=kwargs,
        where=where,
    )

run_daily(func: JobCallable, at: str = '00:00', *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> Coroutine[Any, Any, ScheduledJob]

Schedule a job to run once per day at a fixed wall-clock time.

Must be awaited. Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

Uses a cron-based trigger internally to ensure DST-correct, wall-clock-aligned scheduling. This avoids the 24-hour drift bug of interval-based daily scheduling.

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
at	str	Target wall-clock time in "HH:MM" format (default "00:00").	'00:00'
name	str	Required stable name for the job.	required
group	str | None	Optional group name.	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. See schedule() for details.	None
timeout	float | None	Per-job timeout in seconds. See schedule() for details.	None
timeout_disabled	bool	Disable timeout enforcement. See schedule() for details.	False
mode	ExecutionMode | str | None	Overlap behavior when a prior invocation is still running as the next tick becomes due. See schedule() for the four values, tier-aware default, and string coercion rules.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. See schedule() for details.	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) gating execution. See schedule() for details.	None

Returns:

Type	Description
Coroutine[Any, Any, ScheduledJob]	The scheduled job.

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

def run_daily(
    self,
    func: "JobCallable",
    at: str = "00:00",
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> "Coroutine[Any, Any, ScheduledJob]":
    """Schedule a job to run once per day at a fixed wall-clock time.

    Must be awaited. Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    Uses a cron-based trigger internally to ensure DST-correct, wall-clock-aligned
    scheduling. This avoids the 24-hour drift bug of interval-based daily scheduling.

    Args:
        func: The function to run.
        at: Target wall-clock time in ``"HH:MM"`` format (default ``"00:00"``).
        name: Required stable name for the job.
        group: Optional group name.
        jitter: Optional seconds of random offset to apply at enqueue time.
            See ``schedule()`` for details.
        timeout: Per-job timeout in seconds. See ``schedule()`` for details.
        timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
        mode: Overlap behavior when a prior invocation is still running as the next
            tick becomes due. See ``schedule()`` for the four values, tier-aware
            default, and string coercion rules.
        on_error: Optional per-job error handler. See ``schedule()`` for details.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) gating execution.
            See ``schedule()`` for details.

    Returns:
        The scheduled job.
    """
    trigger = Daily(at=at)
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at add_job (the true primary). See design/071.
    return self.schedule(
        func,
        trigger,
        name=name,
        group=group,
        jitter=jitter,
        timeout=timeout,
        timeout_disabled=timeout_disabled,
        mode=mode,
        on_error=on_error,
        if_exists=if_exists,
        args=args,
        kwargs=kwargs,
        where=where,
    )

run_cron(func: JobCallable, expression: str, *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> Coroutine[Any, Any, ScheduledJob]

Schedule a job using a cron expression.

Must be awaited. Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

Accepts both 5-field (standard Unix cron: minute hour dom month dow) and 6-field expressions (seconds appended as a 6th field per croniter convention: minute hour dom month dow second).

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
expression	str	A valid 5- or 6-field cron expression.	required
name	str	Required stable name for the job.	required
group	str | None	Optional group name.	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. See schedule() for details.	None
timeout	float | None	Per-job timeout in seconds. See schedule() for details.	None
timeout_disabled	bool	Disable timeout enforcement. See schedule() for details.	False
mode	ExecutionMode | str | None	Overlap behavior when a prior invocation is still running as the next tick becomes due. See schedule() for the four values, tier-aware default, and string coercion rules.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. See schedule() for details.	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) gating execution. See schedule() for details.	None

Returns:

Type	Description
Coroutine[Any, Any, ScheduledJob]	The scheduled job.

Raises:

Type	Description
ValueError	If the cron expression is syntactically invalid.

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

def run_cron(
    self,
    func: "JobCallable",
    expression: str,
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> "Coroutine[Any, Any, ScheduledJob]":
    """Schedule a job using a cron expression.

    Must be awaited. Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    Accepts both 5-field (standard Unix cron: ``minute hour dom month dow``)
    and 6-field expressions (seconds appended as a 6th field per croniter
    convention: ``minute hour dom month dow second``).

    Args:
        func: The function to run.
        expression: A valid 5- or 6-field cron expression.
        name: Required stable name for the job.
        group: Optional group name.
        jitter: Optional seconds of random offset to apply at enqueue time.
            See ``schedule()`` for details.
        timeout: Per-job timeout in seconds. See ``schedule()`` for details.
        timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
        mode: Overlap behavior when a prior invocation is still running as the next
            tick becomes due. See ``schedule()`` for the four values, tier-aware
            default, and string coercion rules.
        on_error: Optional per-job error handler. See ``schedule()`` for details.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) gating execution.
            See ``schedule()`` for details.

    Returns:
        The scheduled job.

    Raises:
        ValueError: If the cron expression is syntactically invalid.
    """
    trigger = Cron(expression)
    # Shape B delegate — returns the callee's handle directly (no await, no second guard_await).
    # The single guard_await lives at add_job (the true primary). See design/071.
    return self.schedule(
        func,
        trigger,
        name=name,
        group=group,
        jitter=jitter,
        timeout=timeout,
        timeout_disabled=timeout_disabled,
        mode=mode,
        on_error=on_error,
        if_exists=if_exists,
        args=args,
        kwargs=kwargs,
        where=where,
    )

SchedulerSyncFacade

Bases: Resource

Synchronous facade for the scheduler.

This class provides synchronous methods that wrap the asynchronous scheduling methods of the Scheduler class, allowing jobs to be scheduled from synchronous code (for example, an AppSync lifecycle hook running in a worker thread).

These methods must not be called from within the event loop; doing so raises a RuntimeError. Use the asynchronous methods on Scheduler directly when operating within an event loop.

Source code in src/hassette/scheduler/sync.py

class SchedulerSyncFacade(Resource):
    """Synchronous facade for the scheduler.

    This class provides synchronous methods that wrap the asynchronous scheduling methods of
    the Scheduler class, allowing jobs to be scheduled from synchronous code (for example, an
    ``AppSync`` lifecycle hook running in a worker thread).

    These methods must not be called from within the event loop; doing so raises a RuntimeError.
    Use the asynchronous methods on ``Scheduler`` directly when operating within an event loop.
    """

    _scheduler: "Scheduler"

    def __init__(self, hassette: "Hassette", *, scheduler: "Scheduler", parent: Resource | None = None) -> None:
        super().__init__(hassette, parent=parent)
        self._scheduler = scheduler

    async def on_initialize(self) -> None:
        mark_ready(self, reason="Synchronous Scheduler facade initialized")

    @property
    def config_log_level(self) -> LOG_LEVEL_TYPE:
        return self.hassette.config.logging.scheduler_service

    def add_job(self, job: "ScheduledJob", *, if_exists: IfExistsPolicy = "error") -> ScheduledJob:
        """Add a job to the scheduler.

        Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        DB registration completes inline — ``job.db_id`` is set before this
        method returns, eliminating the window where a job fires with
        ``db_id=None``.

        Args:
            job: The job to add.
            if_exists: Behavior when a job with the same name already exists.
                ``"error"`` (default) raises ``ValueError``.
                ``"skip"`` returns the existing job if it matches; raises
                ``ValueError`` if the name matches but the configuration differs.
                ``"replace"`` cancels the existing job (recording it as cancelled
                in telemetry) and registers the new job in its place.

        Returns:
            The added job, or the existing job when ``if_exists="skip"`` and a
            matching job is already registered. ``job.db_id`` is a valid
            integer immediately on return.

        Raises:
            TypeError: If job is not a ScheduledJob.
            SchedulerNameRequiredError: If ``job.name`` is empty.
            ValueError: If a job with the same name already exists and either
                ``if_exists="error"`` or the existing job's configuration differs.
        """
        return self.task_bucket.run_sync(self._scheduler.add_job(job, if_exists=if_exists))

    def schedule(
        self,
        func: "JobCallable",
        trigger: "TriggerProtocol",
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> ScheduledJob:
        """Schedule a job using a trigger object.

        Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        This is the primary entry point for scheduling. All convenience methods
        (``run_in``, ``run_every``, ``run_daily``, etc.) delegate here.

        Args:
            func: The function to run.
            trigger: A trigger object implementing ``TriggerProtocol``. Determines
                both the first run time and subsequent recurrences.
            name: Required stable name for the job. Used for uniqueness validation
                within this scheduler instance and for logging/telemetry.
            group: Optional group name for bulk management (see ``cancel_group``).
            jitter: Optional seconds of random offset to apply at enqueue time.
                Jitter is applied via ``SchedulerService.apply_jitter_to_heap`` on enqueue.
                See the ``fire_at`` field on ``ScheduledJob``.
            timeout: Per-job timeout in seconds. ``None`` uses the global default.
                A positive ``float`` overrides the default.
            timeout_disabled: When ``True``, timeout enforcement is disabled for this
                job regardless of the global default.
            mode: Overlap behavior when a prior invocation is still running as the next
                occurrence becomes due. One of ``"single"`` (skip the re-fire), ``"restart"``
                (cancel the running invocation and start fresh), ``"queued"`` (serialize
                re-fires in arrival order, up to the cap), or ``"parallel"`` (run concurrently).
                A raw string is coerced to ``ExecutionMode``; an invalid string raises
                ``ValueError`` naming the valid values. Omitted (``None``) resolves
                tier-aware: ``"single"`` for app-tier jobs, ``"parallel"`` for framework-tier
                jobs. Accepted on one-shot schedules (``run_in``/``run_once``) for API
                uniformity, but has no overlap effect since one-shots never re-fire.
            on_error: Optional per-job error handler. When set, this handler is
                invoked if the job raises an exception (including ``TimeoutError``,
                but excluding ``CancelledError``). Overrides the app-level handler
                set via ``on_error()``.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) evaluated at dispatch
                time, before the handler runs. Predicate signatures are inspected via
                the shared DI layer (``hassette.di``): a parameter annotated as
                ``ScheduledJob`` receives the job instance at dispatch time; unannotated
                predicates are called with zero arguments. A sequence is collapsed into
                a single combinator that ANDs all members — each member keeps the
                single-predicate contract. Predicates must be synchronous; async
                callables raise ``TypeError``. When the predicate returns ``False``,
                the handler does not run and a ``'skipped'`` execution is recorded.
                When the predicate raises, the handler does not run, an ``'error'``
                execution is recorded, and the job's ``on_error`` handler is invoked.

        Returns:
            The scheduled job. ``job.db_id`` is a valid integer immediately on return.

        Raises:
            TypeError: If ``trigger`` does not implement ``TriggerProtocol``, or if
                ``where`` is (or contains) an async callable.
            SchedulerNameRequiredError: If ``name`` is empty.
            DependencyInjectionError: If a predicate's signature is incompatible with
                DI (e.g. ``*args`` or positional-only parameters).
        """
        return self.task_bucket.run_sync(
            self._scheduler.schedule(
                func,
                trigger,
                name=name,
                group=group,
                jitter=jitter,
                timeout=timeout,
                timeout_disabled=timeout_disabled,
                mode=mode,
                on_error=on_error,
                if_exists=if_exists,
                args=args,
                kwargs=kwargs,
                where=where,
            )
        )

    def run_in(
        self,
        func: "JobCallable",
        delay: float,
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> ScheduledJob:
        """Schedule a job to run after a fixed delay (one-shot).

        Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        Args:
            func: The function to run.
            delay: The delay in seconds before running the job.
            name: Required stable name for the job.
            group: Optional group name.
            jitter: Optional seconds of random offset to apply at enqueue time.
                See ``schedule()`` for details.
            timeout: Per-job timeout in seconds. See ``schedule()`` for details.
            timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
            mode: Overlap mode. Accepted for API uniformity; has no overlap effect for
                one-shot jobs since they never re-fire. See ``schedule()`` for the four
                values, tier-aware default, and string coercion rules.
            on_error: Optional per-job error handler. See ``schedule()`` for details.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) gating execution.
                See ``schedule()`` for details.

        Returns:
            The scheduled job.
        """
        return self.task_bucket.run_sync(
            self._scheduler.run_in(
                func,
                delay,
                name=name,
                group=group,
                jitter=jitter,
                timeout=timeout,
                timeout_disabled=timeout_disabled,
                mode=mode,
                on_error=on_error,
                if_exists=if_exists,
                args=args,
                kwargs=kwargs,
                where=where,
            )
        )

    def run_once(
        self,
        func: "JobCallable",
        at: str | ZonedDateTime,
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        if_past: Literal["tomorrow", "error"] = "tomorrow",
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> ScheduledJob:
        """Schedule a job to run once at a specific wall-clock time (one-shot).

        Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        Args:
            func: The function to run.
            at: Target time. A ``"HH:MM"`` string (today in the configured
                timezone — falls back to the process timezone when unset —
                or tomorrow if already past) or a ``ZonedDateTime``.
            name: Required stable name for the job.
            group: Optional group name.
            jitter: Optional seconds of random offset to apply at enqueue time.
                See ``schedule()`` for details.
            timeout: Per-job timeout in seconds. See ``schedule()`` for details.
            timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
            if_past: Behaviour when the target time is in the past at construction
                time. ``"tomorrow"`` (default) defers by one day. ``"error"`` raises
                ``ValueError``. For ``ZonedDateTime`` inputs, ``if_past`` has no
                effect — the job always fires immediately if the instant is in the past.
            mode: Overlap mode. Accepted for API uniformity; has no overlap effect for
                one-shot jobs since they never re-fire. See ``schedule()`` for the four
                values, tier-aware default, and string coercion rules.
            on_error: Optional per-job error handler. See ``schedule()`` for details.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) gating execution.
                See ``schedule()`` for details.

        Returns:
            The scheduled job.
        """
        return self.task_bucket.run_sync(
            self._scheduler.run_once(
                func,
                at,
                name=name,
                group=group,
                jitter=jitter,
                timeout=timeout,
                timeout_disabled=timeout_disabled,
                if_past=if_past,
                mode=mode,
                on_error=on_error,
                if_exists=if_exists,
                args=args,
                kwargs=kwargs,
                where=where,
            )
        )

    def run_every(
        self,
        func: "JobCallable",
        hours: float = 0,
        minutes: float = 0,
        seconds: float = 0,
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> ScheduledJob:
        """Schedule a job to run at a fixed interval.

        Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        Args:
            func: The function to run.
            hours: Interval hours component.
            minutes: Interval minutes component.
            seconds: Interval seconds component.
            name: Required stable name for the job.
            group: Optional group name.
            jitter: Optional seconds of random offset to apply at enqueue time.
                See ``schedule()`` for details.
            timeout: Per-job timeout in seconds. See ``schedule()`` for details.
            timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
            mode: Overlap behavior when a prior invocation is still running as the next
                tick becomes due. See ``schedule()`` for the four values, tier-aware
                default, and string coercion rules.
            on_error: Optional per-job error handler. See ``schedule()`` for details.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) gating execution.
                See ``schedule()`` for details.

        Returns:
            The scheduled job.
        """
        return self.task_bucket.run_sync(
            self._scheduler.run_every(
                func,
                hours,
                minutes,
                seconds,
                name=name,
                group=group,
                jitter=jitter,
                timeout=timeout,
                timeout_disabled=timeout_disabled,
                mode=mode,
                on_error=on_error,
                if_exists=if_exists,
                args=args,
                kwargs=kwargs,
                where=where,
            )
        )

    def run_minutely(
        self,
        func: "JobCallable",
        minutes: int = 1,
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> ScheduledJob:
        """Schedule a job to run every N minutes.

        Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        Args:
            func: The function to run.
            minutes: The minute interval (must be >= 1).
            name: Required stable name for the job.
            group: Optional group name.
            jitter: Optional seconds of random offset to apply at enqueue time.
                See ``schedule()`` for details.
            timeout: Per-job timeout in seconds. See ``schedule()`` for details.
            timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
            mode: Overlap behavior when a prior invocation is still running as the next
                tick becomes due. See ``schedule()`` for the four values, tier-aware
                default, and string coercion rules.
            on_error: Optional per-job error handler. See ``schedule()`` for details.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) gating execution.
                See ``schedule()`` for details.

        Returns:
            The scheduled job.
        """
        return self.task_bucket.run_sync(
            self._scheduler.run_minutely(
                func,
                minutes,
                name=name,
                group=group,
                jitter=jitter,
                timeout=timeout,
                timeout_disabled=timeout_disabled,
                mode=mode,
                on_error=on_error,
                if_exists=if_exists,
                args=args,
                kwargs=kwargs,
                where=where,
            )
        )

    def run_hourly(
        self,
        func: "JobCallable",
        hours: int = 1,
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> ScheduledJob:
        """Schedule a job to run every N hours.

        Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        Args:
            func: The function to run.
            hours: The hour interval (must be >= 1).
            name: Required stable name for the job.
            group: Optional group name.
            jitter: Optional seconds of random offset to apply at enqueue time.
                See ``schedule()`` for details.
            timeout: Per-job timeout in seconds. See ``schedule()`` for details.
            timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
            mode: Overlap behavior when a prior invocation is still running as the next
                tick becomes due. See ``schedule()`` for the four values, tier-aware
                default, and string coercion rules.
            on_error: Optional per-job error handler. See ``schedule()`` for details.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) gating execution.
                See ``schedule()`` for details.

        Returns:
            The scheduled job.
        """
        return self.task_bucket.run_sync(
            self._scheduler.run_hourly(
                func,
                hours,
                name=name,
                group=group,
                jitter=jitter,
                timeout=timeout,
                timeout_disabled=timeout_disabled,
                mode=mode,
                on_error=on_error,
                if_exists=if_exists,
                args=args,
                kwargs=kwargs,
                where=where,
            )
        )

    def run_daily(
        self,
        func: "JobCallable",
        at: str = "00:00",
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> ScheduledJob:
        """Schedule a job to run once per day at a fixed wall-clock time.

        Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        Uses a cron-based trigger internally to ensure DST-correct, wall-clock-aligned
        scheduling. This avoids the 24-hour drift bug of interval-based daily scheduling.

        Args:
            func: The function to run.
            at: Target wall-clock time in ``"HH:MM"`` format (default ``"00:00"``).
            name: Required stable name for the job.
            group: Optional group name.
            jitter: Optional seconds of random offset to apply at enqueue time.
                See ``schedule()`` for details.
            timeout: Per-job timeout in seconds. See ``schedule()`` for details.
            timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
            mode: Overlap behavior when a prior invocation is still running as the next
                tick becomes due. See ``schedule()`` for the four values, tier-aware
                default, and string coercion rules.
            on_error: Optional per-job error handler. See ``schedule()`` for details.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) gating execution.
                See ``schedule()`` for details.

        Returns:
            The scheduled job.
        """
        return self.task_bucket.run_sync(
            self._scheduler.run_daily(
                func,
                at,
                name=name,
                group=group,
                jitter=jitter,
                timeout=timeout,
                timeout_disabled=timeout_disabled,
                mode=mode,
                on_error=on_error,
                if_exists=if_exists,
                args=args,
                kwargs=kwargs,
                where=where,
            )
        )

    def run_cron(
        self,
        func: "JobCallable",
        expression: str,
        *,
        name: str,
        group: str | None = None,
        jitter: float | None = None,
        timeout: float | None = None,
        timeout_disabled: bool = False,
        mode: "ExecutionMode | str | None" = None,
        on_error: "SchedulerErrorHandlerType | None" = None,
        if_exists: IfExistsPolicy = "error",
        args: tuple[Any, ...] | None = None,
        kwargs: Mapping[str, Any] | None = None,
        where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
    ) -> ScheduledJob:
        """Schedule a job using a cron expression.

        Scheduling completes before the call returns.
        ``job.db_id`` is a valid integer immediately on return.

        Accepts both 5-field (standard Unix cron: ``minute hour dom month dow``)
        and 6-field expressions (seconds appended as a 6th field per croniter
        convention: ``minute hour dom month dow second``).

        Args:
            func: The function to run.
            expression: A valid 5- or 6-field cron expression.
            name: Required stable name for the job.
            group: Optional group name.
            jitter: Optional seconds of random offset to apply at enqueue time.
                See ``schedule()`` for details.
            timeout: Per-job timeout in seconds. See ``schedule()`` for details.
            timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
            mode: Overlap behavior when a prior invocation is still running as the next
                tick becomes due. See ``schedule()`` for the four values, tier-aware
                default, and string coercion rules.
            on_error: Optional per-job error handler. See ``schedule()`` for details.
            if_exists: Behavior when a job with the same name already exists.
                See :meth:`add_job` for details.
            args: Positional arguments to pass to the callable when it executes.
            kwargs: Keyword arguments to pass to the callable when it executes.
            where: Optional predicate (or sequence of predicates) gating execution.
                See ``schedule()`` for details.

        Returns:
            The scheduled job.

        Raises:
            ValueError: If the cron expression is syntactically invalid.
        """
        return self.task_bucket.run_sync(
            self._scheduler.run_cron(
                func,
                expression,
                name=name,
                group=group,
                jitter=jitter,
                timeout=timeout,
                timeout_disabled=timeout_disabled,
                mode=mode,
                on_error=on_error,
                if_exists=if_exists,
                args=args,
                kwargs=kwargs,
                where=where,
            )
        )

    def on_error(self, handler: "SchedulerErrorHandlerType") -> None:
        """Register an app-level error handler for this scheduler.

        The handler is called when any job on this scheduler raises an exception
        (including ``TimeoutError``) and the job does not have its own
        per-registration error handler.

        This is an app-level fallback — it is resolved at dispatch time, not at job
        registration time. A later call to ``on_error()`` replaces any previously
        registered handler.

        Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near
        app shutdown may be cancelled before they complete. Do not rely on error handlers
        for delivery-critical alerting during system teardown.

        Args:
            handler: A sync or async callable that accepts a
                :class:`~hassette.scheduler.error_context.SchedulerErrorContext`.
        """
        return self._scheduler.on_error(handler)

    def cancel_job(self, job: "ScheduledJob") -> None:
        """Cancel an individual job and persist the cancellation to the database.

        Idempotent: a second cancel on the same job is a silent no-op. Raises
        ``ValueError`` if the job belongs to a different scheduler instance.
        Spawns a durable ``mark_job_cancelled`` DB write (when ``db_id`` is set),
        dequeues the job from the service, and sets ``job._dequeued = True``.

        Must NOT call ``job.cancel()`` internally — that delegates back here and
        would cause infinite recursion.

        Args:
            job: The job to cancel.

        Raises:
            ValueError: If the job belongs to a different scheduler instance.
        """
        return self._scheduler.cancel_job(job)

    def cancel_group(self, group: str) -> None:
        """Cancel all jobs in the given group.

        Delegates to ``cancel_job`` per-member, which handles the DB write,
        dequeue, and ``_dequeued`` flag. Dict cleanup (``_jobs_by_group`` and
        ``_jobs_by_name``) is handled by the ``_on_job_removed`` callback
        fired by ``scheduler_service.dequeue_job``. No-op if the group does
        not exist.

        Args:
            group: The group name to cancel.
        """
        return self._scheduler.cancel_group(group)

    def list_jobs(self, group: str | None = None) -> list["ScheduledJob"]:
        """Return all or group-filtered jobs.

        Args:
            group: If provided, return only jobs in this group.
                If ``None`` (default), return all jobs.

        Returns:
            List of ScheduledJob instances.
        """
        return self._scheduler.list_jobs(group)

add_job(job: ScheduledJob, *, if_exists: IfExistsPolicy = 'error') -> ScheduledJob

Add a job to the scheduler.

Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

DB registration completes inline — job.db_id is set before this method returns, eliminating the window where a job fires with db_id=None.

Parameters:

Name	Type	Description	Default
job	ScheduledJob	The job to add.	required
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. "error" (default) raises ValueError. "skip" returns the existing job if it matches; raises ValueError if the name matches but the configuration differs. "replace" cancels the existing job (recording it as cancelled in telemetry) and registers the new job in its place.	'error'

Returns:

Type	Description
ScheduledJob	The added job, or the existing job when if_exists="skip" and a
ScheduledJob	matching job is already registered. job.db_id is a valid
ScheduledJob	integer immediately on return.

Raises:

Type	Description
TypeError	If job is not a ScheduledJob.
SchedulerNameRequiredError	If job.name is empty.
ValueError	If a job with the same name already exists and either if_exists="error" or the existing job's configuration differs.

Source code in src/hassette/scheduler/sync.py

def add_job(self, job: "ScheduledJob", *, if_exists: IfExistsPolicy = "error") -> ScheduledJob:
    """Add a job to the scheduler.

    Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    DB registration completes inline — ``job.db_id`` is set before this
    method returns, eliminating the window where a job fires with
    ``db_id=None``.

    Args:
        job: The job to add.
        if_exists: Behavior when a job with the same name already exists.
            ``"error"`` (default) raises ``ValueError``.
            ``"skip"`` returns the existing job if it matches; raises
            ``ValueError`` if the name matches but the configuration differs.
            ``"replace"`` cancels the existing job (recording it as cancelled
            in telemetry) and registers the new job in its place.

    Returns:
        The added job, or the existing job when ``if_exists="skip"`` and a
        matching job is already registered. ``job.db_id`` is a valid
        integer immediately on return.

    Raises:
        TypeError: If job is not a ScheduledJob.
        SchedulerNameRequiredError: If ``job.name`` is empty.
        ValueError: If a job with the same name already exists and either
            ``if_exists="error"`` or the existing job's configuration differs.
    """
    return self.task_bucket.run_sync(self._scheduler.add_job(job, if_exists=if_exists))

schedule(func: JobCallable, trigger: TriggerProtocol, *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> ScheduledJob

Schedule a job using a trigger object.

Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

This is the primary entry point for scheduling. All convenience methods (run_in, run_every, run_daily, etc.) delegate here.

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
trigger	TriggerProtocol	A trigger object implementing TriggerProtocol. Determines both the first run time and subsequent recurrences.	required
name	str	Required stable name for the job. Used for uniqueness validation within this scheduler instance and for logging/telemetry.	required
group	str | None	Optional group name for bulk management (see cancel_group).	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. Jitter is applied via SchedulerService.apply_jitter_to_heap on enqueue. See the fire_at field on ScheduledJob.	None
timeout	float | None	Per-job timeout in seconds. None uses the global default. A positive float overrides the default.	None
timeout_disabled	bool	When True, timeout enforcement is disabled for this job regardless of the global default.	False
mode	ExecutionMode | str | None	Overlap behavior when a prior invocation is still running as the next occurrence becomes due. One of "single" (skip the re-fire), "restart" (cancel the running invocation and start fresh), "queued" (serialize re-fires in arrival order, up to the cap), or "parallel" (run concurrently). A raw string is coerced to ExecutionMode; an invalid string raises ValueError naming the valid values. Omitted (None) resolves tier-aware: "single" for app-tier jobs, "parallel" for framework-tier jobs. Accepted on one-shot schedules (run_in/run_once) for API uniformity, but has no overlap effect since one-shots never re-fire.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. When set, this handler is invoked if the job raises an exception (including TimeoutError, but excluding CancelledError). Overrides the app-level handler set via on_error().	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) evaluated at dispatch time, before the handler runs. Predicate signatures are inspected via the shared DI layer (hassette.di): a parameter annotated as ScheduledJob receives the job instance at dispatch time; unannotated predicates are called with zero arguments. A sequence is collapsed into a single combinator that ANDs all members — each member keeps the single-predicate contract. Predicates must be synchronous; async callables raise TypeError. When the predicate returns False, the handler does not run and a 'skipped' execution is recorded. When the predicate raises, the handler does not run, an 'error' execution is recorded, and the job's on_error handler is invoked.	None

Returns:

Type	Description
ScheduledJob	The scheduled job. job.db_id is a valid integer immediately on return.

Raises:

Type	Description
TypeError	If trigger does not implement TriggerProtocol, or if where is (or contains) an async callable.
SchedulerNameRequiredError	If name is empty.
DependencyInjectionError	If a predicate's signature is incompatible with DI (e.g. *args or positional-only parameters).

Source code in src/hassette/scheduler/sync.py

def schedule(
    self,
    func: "JobCallable",
    trigger: "TriggerProtocol",
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> ScheduledJob:
    """Schedule a job using a trigger object.

    Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    This is the primary entry point for scheduling. All convenience methods
    (``run_in``, ``run_every``, ``run_daily``, etc.) delegate here.

    Args:
        func: The function to run.
        trigger: A trigger object implementing ``TriggerProtocol``. Determines
            both the first run time and subsequent recurrences.
        name: Required stable name for the job. Used for uniqueness validation
            within this scheduler instance and for logging/telemetry.
        group: Optional group name for bulk management (see ``cancel_group``).
        jitter: Optional seconds of random offset to apply at enqueue time.
            Jitter is applied via ``SchedulerService.apply_jitter_to_heap`` on enqueue.
            See the ``fire_at`` field on ``ScheduledJob``.
        timeout: Per-job timeout in seconds. ``None`` uses the global default.
            A positive ``float`` overrides the default.
        timeout_disabled: When ``True``, timeout enforcement is disabled for this
            job regardless of the global default.
        mode: Overlap behavior when a prior invocation is still running as the next
            occurrence becomes due. One of ``"single"`` (skip the re-fire), ``"restart"``
            (cancel the running invocation and start fresh), ``"queued"`` (serialize
            re-fires in arrival order, up to the cap), or ``"parallel"`` (run concurrently).
            A raw string is coerced to ``ExecutionMode``; an invalid string raises
            ``ValueError`` naming the valid values. Omitted (``None``) resolves
            tier-aware: ``"single"`` for app-tier jobs, ``"parallel"`` for framework-tier
            jobs. Accepted on one-shot schedules (``run_in``/``run_once``) for API
            uniformity, but has no overlap effect since one-shots never re-fire.
        on_error: Optional per-job error handler. When set, this handler is
            invoked if the job raises an exception (including ``TimeoutError``,
            but excluding ``CancelledError``). Overrides the app-level handler
            set via ``on_error()``.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) evaluated at dispatch
            time, before the handler runs. Predicate signatures are inspected via
            the shared DI layer (``hassette.di``): a parameter annotated as
            ``ScheduledJob`` receives the job instance at dispatch time; unannotated
            predicates are called with zero arguments. A sequence is collapsed into
            a single combinator that ANDs all members — each member keeps the
            single-predicate contract. Predicates must be synchronous; async
            callables raise ``TypeError``. When the predicate returns ``False``,
            the handler does not run and a ``'skipped'`` execution is recorded.
            When the predicate raises, the handler does not run, an ``'error'``
            execution is recorded, and the job's ``on_error`` handler is invoked.

    Returns:
        The scheduled job. ``job.db_id`` is a valid integer immediately on return.

    Raises:
        TypeError: If ``trigger`` does not implement ``TriggerProtocol``, or if
            ``where`` is (or contains) an async callable.
        SchedulerNameRequiredError: If ``name`` is empty.
        DependencyInjectionError: If a predicate's signature is incompatible with
            DI (e.g. ``*args`` or positional-only parameters).
    """
    return self.task_bucket.run_sync(
        self._scheduler.schedule(
            func,
            trigger,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )
    )

run_in(func: JobCallable, delay: float, *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> ScheduledJob

Schedule a job to run after a fixed delay (one-shot).

Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
delay	float	The delay in seconds before running the job.	required
name	str	Required stable name for the job.	required
group	str | None	Optional group name.	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. See schedule() for details.	None
timeout	float | None	Per-job timeout in seconds. See schedule() for details.	None
timeout_disabled	bool	Disable timeout enforcement. See schedule() for details.	False
mode	ExecutionMode | str | None	Overlap mode. Accepted for API uniformity; has no overlap effect for one-shot jobs since they never re-fire. See schedule() for the four values, tier-aware default, and string coercion rules.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. See schedule() for details.	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) gating execution. See schedule() for details.	None

Returns:

Type	Description
ScheduledJob	The scheduled job.

Source code in src/hassette/scheduler/sync.py

def run_in(
    self,
    func: "JobCallable",
    delay: float,
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> ScheduledJob:
    """Schedule a job to run after a fixed delay (one-shot).

    Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    Args:
        func: The function to run.
        delay: The delay in seconds before running the job.
        name: Required stable name for the job.
        group: Optional group name.
        jitter: Optional seconds of random offset to apply at enqueue time.
            See ``schedule()`` for details.
        timeout: Per-job timeout in seconds. See ``schedule()`` for details.
        timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
        mode: Overlap mode. Accepted for API uniformity; has no overlap effect for
            one-shot jobs since they never re-fire. See ``schedule()`` for the four
            values, tier-aware default, and string coercion rules.
        on_error: Optional per-job error handler. See ``schedule()`` for details.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) gating execution.
            See ``schedule()`` for details.

    Returns:
        The scheduled job.
    """
    return self.task_bucket.run_sync(
        self._scheduler.run_in(
            func,
            delay,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )
    )

run_once(func: JobCallable, at: str | ZonedDateTime, *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, if_past: Literal['tomorrow', 'error'] = 'tomorrow', mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> ScheduledJob

Schedule a job to run once at a specific wall-clock time (one-shot).

Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
at	str | ZonedDateTime	Target time. A "HH:MM" string (today in the configured timezone — falls back to the process timezone when unset — or tomorrow if already past) or a ZonedDateTime.	required
name	str	Required stable name for the job.	required
group	str | None	Optional group name.	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. See schedule() for details.	None
timeout	float | None	Per-job timeout in seconds. See schedule() for details.	None
timeout_disabled	bool	Disable timeout enforcement. See schedule() for details.	False
if_past	Literal['tomorrow', 'error']	Behaviour when the target time is in the past at construction time. "tomorrow" (default) defers by one day. "error" raises ValueError. For ZonedDateTime inputs, if_past has no effect — the job always fires immediately if the instant is in the past.	'tomorrow'
mode	ExecutionMode | str | None	Overlap mode. Accepted for API uniformity; has no overlap effect for one-shot jobs since they never re-fire. See schedule() for the four values, tier-aware default, and string coercion rules.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. See schedule() for details.	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) gating execution. See schedule() for details.	None

Returns:

Type	Description
ScheduledJob	The scheduled job.

Source code in src/hassette/scheduler/sync.py

def run_once(
    self,
    func: "JobCallable",
    at: str | ZonedDateTime,
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    if_past: Literal["tomorrow", "error"] = "tomorrow",
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> ScheduledJob:
    """Schedule a job to run once at a specific wall-clock time (one-shot).

    Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    Args:
        func: The function to run.
        at: Target time. A ``"HH:MM"`` string (today in the configured
            timezone — falls back to the process timezone when unset —
            or tomorrow if already past) or a ``ZonedDateTime``.
        name: Required stable name for the job.
        group: Optional group name.
        jitter: Optional seconds of random offset to apply at enqueue time.
            See ``schedule()`` for details.
        timeout: Per-job timeout in seconds. See ``schedule()`` for details.
        timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
        if_past: Behaviour when the target time is in the past at construction
            time. ``"tomorrow"`` (default) defers by one day. ``"error"`` raises
            ``ValueError``. For ``ZonedDateTime`` inputs, ``if_past`` has no
            effect — the job always fires immediately if the instant is in the past.
        mode: Overlap mode. Accepted for API uniformity; has no overlap effect for
            one-shot jobs since they never re-fire. See ``schedule()`` for the four
            values, tier-aware default, and string coercion rules.
        on_error: Optional per-job error handler. See ``schedule()`` for details.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) gating execution.
            See ``schedule()`` for details.

    Returns:
        The scheduled job.
    """
    return self.task_bucket.run_sync(
        self._scheduler.run_once(
            func,
            at,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            if_past=if_past,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )
    )

run_every(func: JobCallable, hours: float = 0, minutes: float = 0, seconds: float = 0, *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> ScheduledJob

Schedule a job to run at a fixed interval.

Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
hours	float	Interval hours component.	0
minutes	float	Interval minutes component.	0
seconds	float	Interval seconds component.	0
name	str	Required stable name for the job.	required
group	str | None	Optional group name.	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. See schedule() for details.	None
timeout	float | None	Per-job timeout in seconds. See schedule() for details.	None
timeout_disabled	bool	Disable timeout enforcement. See schedule() for details.	False
mode	ExecutionMode | str | None	Overlap behavior when a prior invocation is still running as the next tick becomes due. See schedule() for the four values, tier-aware default, and string coercion rules.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. See schedule() for details.	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) gating execution. See schedule() for details.	None

Returns:

Type	Description
ScheduledJob	The scheduled job.

Source code in src/hassette/scheduler/sync.py

def run_every(
    self,
    func: "JobCallable",
    hours: float = 0,
    minutes: float = 0,
    seconds: float = 0,
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> ScheduledJob:
    """Schedule a job to run at a fixed interval.

    Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    Args:
        func: The function to run.
        hours: Interval hours component.
        minutes: Interval minutes component.
        seconds: Interval seconds component.
        name: Required stable name for the job.
        group: Optional group name.
        jitter: Optional seconds of random offset to apply at enqueue time.
            See ``schedule()`` for details.
        timeout: Per-job timeout in seconds. See ``schedule()`` for details.
        timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
        mode: Overlap behavior when a prior invocation is still running as the next
            tick becomes due. See ``schedule()`` for the four values, tier-aware
            default, and string coercion rules.
        on_error: Optional per-job error handler. See ``schedule()`` for details.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) gating execution.
            See ``schedule()`` for details.

    Returns:
        The scheduled job.
    """
    return self.task_bucket.run_sync(
        self._scheduler.run_every(
            func,
            hours,
            minutes,
            seconds,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )
    )

run_minutely(func: JobCallable, minutes: int = 1, *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> ScheduledJob

Schedule a job to run every N minutes.

Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
minutes	int	The minute interval (must be >= 1).	1
name	str	Required stable name for the job.	required
group	str | None	Optional group name.	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. See schedule() for details.	None
timeout	float | None	Per-job timeout in seconds. See schedule() for details.	None
timeout_disabled	bool	Disable timeout enforcement. See schedule() for details.	False
mode	ExecutionMode | str | None	Overlap behavior when a prior invocation is still running as the next tick becomes due. See schedule() for the four values, tier-aware default, and string coercion rules.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. See schedule() for details.	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) gating execution. See schedule() for details.	None

Returns:

Type	Description
ScheduledJob	The scheduled job.

Source code in src/hassette/scheduler/sync.py

def run_minutely(
    self,
    func: "JobCallable",
    minutes: int = 1,
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> ScheduledJob:
    """Schedule a job to run every N minutes.

    Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    Args:
        func: The function to run.
        minutes: The minute interval (must be >= 1).
        name: Required stable name for the job.
        group: Optional group name.
        jitter: Optional seconds of random offset to apply at enqueue time.
            See ``schedule()`` for details.
        timeout: Per-job timeout in seconds. See ``schedule()`` for details.
        timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
        mode: Overlap behavior when a prior invocation is still running as the next
            tick becomes due. See ``schedule()`` for the four values, tier-aware
            default, and string coercion rules.
        on_error: Optional per-job error handler. See ``schedule()`` for details.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) gating execution.
            See ``schedule()`` for details.

    Returns:
        The scheduled job.
    """
    return self.task_bucket.run_sync(
        self._scheduler.run_minutely(
            func,
            minutes,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )
    )

run_hourly(func: JobCallable, hours: int = 1, *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> ScheduledJob

Schedule a job to run every N hours.

Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
hours	int	The hour interval (must be >= 1).	1
name	str	Required stable name for the job.	required
group	str | None	Optional group name.	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. See schedule() for details.	None
timeout	float | None	Per-job timeout in seconds. See schedule() for details.	None
timeout_disabled	bool	Disable timeout enforcement. See schedule() for details.	False
mode	ExecutionMode | str | None	Overlap behavior when a prior invocation is still running as the next tick becomes due. See schedule() for the four values, tier-aware default, and string coercion rules.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. See schedule() for details.	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) gating execution. See schedule() for details.	None

Returns:

Type	Description
ScheduledJob	The scheduled job.

Source code in src/hassette/scheduler/sync.py

def run_hourly(
    self,
    func: "JobCallable",
    hours: int = 1,
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> ScheduledJob:
    """Schedule a job to run every N hours.

    Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    Args:
        func: The function to run.
        hours: The hour interval (must be >= 1).
        name: Required stable name for the job.
        group: Optional group name.
        jitter: Optional seconds of random offset to apply at enqueue time.
            See ``schedule()`` for details.
        timeout: Per-job timeout in seconds. See ``schedule()`` for details.
        timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
        mode: Overlap behavior when a prior invocation is still running as the next
            tick becomes due. See ``schedule()`` for the four values, tier-aware
            default, and string coercion rules.
        on_error: Optional per-job error handler. See ``schedule()`` for details.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) gating execution.
            See ``schedule()`` for details.

    Returns:
        The scheduled job.
    """
    return self.task_bucket.run_sync(
        self._scheduler.run_hourly(
            func,
            hours,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )
    )

run_daily(func: JobCallable, at: str = '00:00', *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> ScheduledJob

Schedule a job to run once per day at a fixed wall-clock time.

Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

Uses a cron-based trigger internally to ensure DST-correct, wall-clock-aligned scheduling. This avoids the 24-hour drift bug of interval-based daily scheduling.

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
at	str	Target wall-clock time in "HH:MM" format (default "00:00").	'00:00'
name	str	Required stable name for the job.	required
group	str | None	Optional group name.	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. See schedule() for details.	None
timeout	float | None	Per-job timeout in seconds. See schedule() for details.	None
timeout_disabled	bool	Disable timeout enforcement. See schedule() for details.	False
mode	ExecutionMode | str | None	Overlap behavior when a prior invocation is still running as the next tick becomes due. See schedule() for the four values, tier-aware default, and string coercion rules.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. See schedule() for details.	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) gating execution. See schedule() for details.	None

Returns:

Type	Description
ScheduledJob	The scheduled job.

Source code in src/hassette/scheduler/sync.py

def run_daily(
    self,
    func: "JobCallable",
    at: str = "00:00",
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> ScheduledJob:
    """Schedule a job to run once per day at a fixed wall-clock time.

    Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    Uses a cron-based trigger internally to ensure DST-correct, wall-clock-aligned
    scheduling. This avoids the 24-hour drift bug of interval-based daily scheduling.

    Args:
        func: The function to run.
        at: Target wall-clock time in ``"HH:MM"`` format (default ``"00:00"``).
        name: Required stable name for the job.
        group: Optional group name.
        jitter: Optional seconds of random offset to apply at enqueue time.
            See ``schedule()`` for details.
        timeout: Per-job timeout in seconds. See ``schedule()`` for details.
        timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
        mode: Overlap behavior when a prior invocation is still running as the next
            tick becomes due. See ``schedule()`` for the four values, tier-aware
            default, and string coercion rules.
        on_error: Optional per-job error handler. See ``schedule()`` for details.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) gating execution.
            See ``schedule()`` for details.

    Returns:
        The scheduled job.
    """
    return self.task_bucket.run_sync(
        self._scheduler.run_daily(
            func,
            at,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )
    )

run_cron(func: JobCallable, expression: str, *, name: str, group: str | None = None, jitter: float | None = None, timeout: float | None = None, timeout_disabled: bool = False, mode: ExecutionMode | str | None = None, on_error: SchedulerErrorHandlerType | None = None, if_exists: IfExistsPolicy = 'error', args: tuple[Any, ...] | None = None, kwargs: Mapping[str, Any] | None = None, where: SchedulerPredicate | Sequence[SchedulerPredicate] | None = None) -> ScheduledJob

Schedule a job using a cron expression.

Scheduling completes before the call returns. job.db_id is a valid integer immediately on return.

Accepts both 5-field (standard Unix cron: minute hour dom month dow) and 6-field expressions (seconds appended as a 6th field per croniter convention: minute hour dom month dow second).

Parameters:

Name	Type	Description	Default
func	JobCallable	The function to run.	required
expression	str	A valid 5- or 6-field cron expression.	required
name	str	Required stable name for the job.	required
group	str | None	Optional group name.	None
jitter	float | None	Optional seconds of random offset to apply at enqueue time. See schedule() for details.	None
timeout	float | None	Per-job timeout in seconds. See schedule() for details.	None
timeout_disabled	bool	Disable timeout enforcement. See schedule() for details.	False
mode	ExecutionMode | str | None	Overlap behavior when a prior invocation is still running as the next tick becomes due. See schedule() for the four values, tier-aware default, and string coercion rules.	None
on_error	SchedulerErrorHandlerType | None	Optional per-job error handler. See schedule() for details.	None
if_exists	IfExistsPolicy	Behavior when a job with the same name already exists. See :meth:add_job for details.	'error'
args	tuple[Any, ...] | None	Positional arguments to pass to the callable when it executes.	None
kwargs	Mapping[str, Any] | None	Keyword arguments to pass to the callable when it executes.	None
where	SchedulerPredicate | Sequence[SchedulerPredicate] | None	Optional predicate (or sequence of predicates) gating execution. See schedule() for details.	None

Returns:

Type	Description
ScheduledJob	The scheduled job.

Raises:

Type	Description
ValueError	If the cron expression is syntactically invalid.

Source code in src/hassette/scheduler/sync.py

def run_cron(
    self,
    func: "JobCallable",
    expression: str,
    *,
    name: str,
    group: str | None = None,
    jitter: float | None = None,
    timeout: float | None = None,
    timeout_disabled: bool = False,
    mode: "ExecutionMode | str | None" = None,
    on_error: "SchedulerErrorHandlerType | None" = None,
    if_exists: IfExistsPolicy = "error",
    args: tuple[Any, ...] | None = None,
    kwargs: Mapping[str, Any] | None = None,
    where: "SchedulerPredicate | Sequence[SchedulerPredicate] | None" = None,
) -> ScheduledJob:
    """Schedule a job using a cron expression.

    Scheduling completes before the call returns.
    ``job.db_id`` is a valid integer immediately on return.

    Accepts both 5-field (standard Unix cron: ``minute hour dom month dow``)
    and 6-field expressions (seconds appended as a 6th field per croniter
    convention: ``minute hour dom month dow second``).

    Args:
        func: The function to run.
        expression: A valid 5- or 6-field cron expression.
        name: Required stable name for the job.
        group: Optional group name.
        jitter: Optional seconds of random offset to apply at enqueue time.
            See ``schedule()`` for details.
        timeout: Per-job timeout in seconds. See ``schedule()`` for details.
        timeout_disabled: Disable timeout enforcement. See ``schedule()`` for details.
        mode: Overlap behavior when a prior invocation is still running as the next
            tick becomes due. See ``schedule()`` for the four values, tier-aware
            default, and string coercion rules.
        on_error: Optional per-job error handler. See ``schedule()`` for details.
        if_exists: Behavior when a job with the same name already exists.
            See :meth:`add_job` for details.
        args: Positional arguments to pass to the callable when it executes.
        kwargs: Keyword arguments to pass to the callable when it executes.
        where: Optional predicate (or sequence of predicates) gating execution.
            See ``schedule()`` for details.

    Returns:
        The scheduled job.

    Raises:
        ValueError: If the cron expression is syntactically invalid.
    """
    return self.task_bucket.run_sync(
        self._scheduler.run_cron(
            func,
            expression,
            name=name,
            group=group,
            jitter=jitter,
            timeout=timeout,
            timeout_disabled=timeout_disabled,
            mode=mode,
            on_error=on_error,
            if_exists=if_exists,
            args=args,
            kwargs=kwargs,
            where=where,
        )
    )

on_error(handler: SchedulerErrorHandlerType) -> None

Register an app-level error handler for this scheduler.

The handler is called when any job on this scheduler raises an exception (including TimeoutError) and the job does not have its own per-registration error handler.

This is an app-level fallback — it is resolved at dispatch time, not at job registration time. A later call to on_error() replaces any previously registered handler.

Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near app shutdown may be cancelled before they complete. Do not rely on error handlers for delivery-critical alerting during system teardown.

Parameters:

Name	Type	Description	Default
handler	SchedulerErrorHandlerType	A sync or async callable that accepts a :class:~hassette.scheduler.error_context.SchedulerErrorContext.	required

Source code in src/hassette/scheduler/sync.py

def on_error(self, handler: "SchedulerErrorHandlerType") -> None:
    """Register an app-level error handler for this scheduler.

    The handler is called when any job on this scheduler raises an exception
    (including ``TimeoutError``) and the job does not have its own
    per-registration error handler.

    This is an app-level fallback — it is resolved at dispatch time, not at job
    registration time. A later call to ``on_error()`` replaces any previously
    registered handler.

    Note: error handlers are spawned as fire-and-forget tasks. Handlers spawned near
    app shutdown may be cancelled before they complete. Do not rely on error handlers
    for delivery-critical alerting during system teardown.

    Args:
        handler: A sync or async callable that accepts a
            :class:`~hassette.scheduler.error_context.SchedulerErrorContext`.
    """
    return self._scheduler.on_error(handler)

cancel_job(job: ScheduledJob) -> None

Cancel an individual job and persist the cancellation to the database.

Idempotent: a second cancel on the same job is a silent no-op. Raises ValueError if the job belongs to a different scheduler instance. Spawns a durable mark_job_cancelled DB write (when db_id is set), dequeues the job from the service, and sets job._dequeued = True.

Must NOT call job.cancel() internally — that delegates back here and would cause infinite recursion.

Parameters:

Name	Type	Description	Default
job	ScheduledJob	The job to cancel.	required

Raises:

Type	Description
ValueError	If the job belongs to a different scheduler instance.

Source code in src/hassette/scheduler/sync.py

def cancel_job(self, job: "ScheduledJob") -> None:
    """Cancel an individual job and persist the cancellation to the database.

    Idempotent: a second cancel on the same job is a silent no-op. Raises
    ``ValueError`` if the job belongs to a different scheduler instance.
    Spawns a durable ``mark_job_cancelled`` DB write (when ``db_id`` is set),
    dequeues the job from the service, and sets ``job._dequeued = True``.

    Must NOT call ``job.cancel()`` internally — that delegates back here and
    would cause infinite recursion.

    Args:
        job: The job to cancel.

    Raises:
        ValueError: If the job belongs to a different scheduler instance.
    """
    return self._scheduler.cancel_job(job)

cancel_group(group: str) -> None

Cancel all jobs in the given group.

Delegates to cancel_job per-member, which handles the DB write, dequeue, and _dequeued flag. Dict cleanup (_jobs_by_group and _jobs_by_name) is handled by the _on_job_removed callback fired by scheduler_service.dequeue_job. No-op if the group does not exist.

Parameters:

Name	Type	Description	Default
group	str	The group name to cancel.	required

Source code in src/hassette/scheduler/sync.py

def cancel_group(self, group: str) -> None:
    """Cancel all jobs in the given group.

    Delegates to ``cancel_job`` per-member, which handles the DB write,
    dequeue, and ``_dequeued`` flag. Dict cleanup (``_jobs_by_group`` and
    ``_jobs_by_name``) is handled by the ``_on_job_removed`` callback
    fired by ``scheduler_service.dequeue_job``. No-op if the group does
    not exist.

    Args:
        group: The group name to cancel.
    """
    return self._scheduler.cancel_group(group)

list_jobs(group: str | None = None) -> list[ScheduledJob]

Return all or group-filtered jobs.

Parameters:

Name	Type	Description	Default
group	str | None	If provided, return only jobs in this group. If None (default), return all jobs.	None

Returns:

Type	Description
list[ScheduledJob]	List of ScheduledJob instances.

Source code in src/hassette/scheduler/sync.py

def list_jobs(self, group: str | None = None) -> list["ScheduledJob"]:
    """Return all or group-filtered jobs.

    Args:
        group: If provided, return only jobs in this group.
            If ``None`` (default), return all jobs.

    Returns:
        List of ScheduledJob instances.
    """
    return self._scheduler.list_jobs(group)

After

One-shot trigger that fires once after a fixed delay.

Accepts seconds, minutes, or a TimeDelta directly.

Parameters:

Name	Type	Description	Default
seconds	float	Delay in seconds.	0
minutes	float	Delay in minutes.	0
timedelta	TimeDelta | None	Delay as a TimeDelta object. Mutually exclusive with seconds/minutes.	None

Example

After(seconds=30) # fires 30 seconds from now After(minutes=5) # fires 5 minutes from now

Source code in src/hassette/scheduler/triggers.py

class After:
    """One-shot trigger that fires once after a fixed delay.

    Accepts seconds, minutes, or a TimeDelta directly.

    Args:
        seconds: Delay in seconds.
        minutes: Delay in minutes.
        timedelta: Delay as a TimeDelta object. Mutually exclusive with seconds/minutes.

    Example:
        After(seconds=30)       # fires 30 seconds from now
        After(minutes=5)        # fires 5 minutes from now
    """

    def __init__(
        self,
        seconds: float = 0,
        minutes: float = 0,
        timedelta: TimeDelta | None = None,
    ) -> None:
        if timedelta is not None:
            self._delay = timedelta
        else:
            self._delay = TimeDelta(seconds=seconds, minutes=minutes)
        if self._delay.total("seconds") <= 0:
            raise ValueError("After trigger delay must be positive")

    def first_run_time(self, current_time: ZonedDateTime) -> ZonedDateTime:
        """Return current_time plus the delay."""
        return current_time.add(seconds=self._delay.total("seconds")).round("second")

    def next_run_time(self, previous_run: ZonedDateTime, current_time: ZonedDateTime) -> None:
        """One-shot trigger; always returns None."""
        return

    def trigger_label(self) -> str:
        return "after"

    def trigger_detail(self) -> str | None:
        return f"{int(self._delay.total('seconds'))}s"

    def trigger_db_type(self) -> Literal["after"]:
        return "after"

    def trigger_id(self) -> str:
        return f"after:{int(self._delay.total('seconds'))}"

first_run_time(current_time: ZonedDateTime) -> ZonedDateTime

Return current_time plus the delay.

Source code in src/hassette/scheduler/triggers.py

def first_run_time(self, current_time: ZonedDateTime) -> ZonedDateTime:
    """Return current_time plus the delay."""
    return current_time.add(seconds=self._delay.total("seconds")).round("second")

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

One-shot trigger; always returns None.

Source code in src/hassette/scheduler/triggers.py

def next_run_time(self, previous_run: ZonedDateTime, current_time: ZonedDateTime) -> None:
    """One-shot trigger; always returns None."""
    return

Cron

Trigger based on an arbitrary cron expression.

Accepts both 5-field (standard Unix cron: minute hour dom month dow) and 6-field expressions (seconds appended as a 6th field per croniter convention: minute hour dom month dow second).

Parameters:

Name	Type	Description	Default
expression	str	A valid 5- or 6-field cron expression.	required

Raises:

Type	Description
ValueError	If the expression is syntactically invalid.

Example

Cron("0 9 * * 1-5") # weekdays at 09:00 Cron("0 9 * * 1-5 0") # weekdays at 09:00:00 (6-field)

Source code in src/hassette/scheduler/triggers.py

class Cron:
    """Trigger based on an arbitrary cron expression.

    Accepts both 5-field (standard Unix cron: ``minute hour dom month dow``)
    and 6-field expressions (seconds appended as a 6th field per croniter
    convention: ``minute hour dom month dow second``).

    Args:
        expression: A valid 5- or 6-field cron expression.

    Raises:
        ValueError: If the expression is syntactically invalid.

    Example:
        Cron("0 9 * * 1-5")    # weekdays at 09:00
        Cron("0 9 * * 1-5 0")  # weekdays at 09:00:00 (6-field)
    """

    def __init__(self, expression: str) -> None:
        self._expression = expression
        try:
            self._cron = CronTrigger(expression)
        except ValueError as exc:
            raise ValueError(f"Invalid cron expression: {expression!r}") from exc

    def first_run_time(self, current_time: ZonedDateTime) -> ZonedDateTime:
        """Return the first cron-grid-aligned run time at or after current_time."""
        return self._cron.first_run_time(current_time)

    def next_run_time(self, previous_run: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime:
        """Return the next cron-grid-aligned run time after previous_run that is later than current_time."""
        return self._cron.next_run_time(previous_run, current_time)

    def trigger_label(self) -> str:
        return "cron"

    def trigger_detail(self) -> str | None:
        return self._expression

    def trigger_db_type(self) -> Literal["cron"]:
        return "cron"

    def trigger_id(self) -> str:
        return f"cron:{self._expression}"

first_run_time(current_time: ZonedDateTime) -> ZonedDateTime

Return the first cron-grid-aligned run time at or after current_time.

Source code in src/hassette/scheduler/triggers.py

def first_run_time(self, current_time: ZonedDateTime) -> ZonedDateTime:
    """Return the first cron-grid-aligned run time at or after current_time."""
    return self._cron.first_run_time(current_time)

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

Return the next cron-grid-aligned run time after previous_run that is later than current_time.

Source code in src/hassette/scheduler/triggers.py

def next_run_time(self, previous_run: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime:
    """Return the next cron-grid-aligned run time after previous_run that is later than current_time."""
    return self._cron.next_run_time(previous_run, current_time)

Daily

Trigger that fires once per day at a fixed wall-clock time.

Internally delegates to a 5-field cron expression to ensure DST-correct, wall-clock-aligned scheduling.

Parameters:

Name	Type	Description	Default
at	str	Target time in "HH:MM" format (e.g. "07:00"), interpreted in the configured timezone (see HassetteConfig.timezone; falls back to the process timezone when unset).	required

Example

Daily(at="07:00") # fires every day at 07:00 wall-clock time

Source code in src/hassette/scheduler/triggers.py

class Daily:
    """Trigger that fires once per day at a fixed wall-clock time.

    Internally delegates to a 5-field cron expression to ensure DST-correct,
    wall-clock-aligned scheduling.

    Args:
        at: Target time in ``"HH:MM"`` format (e.g. ``"07:00"``),
            interpreted in the configured timezone (see
            ``HassetteConfig.timezone``; falls back to the process timezone
            when unset).

    Example:
        Daily(at="07:00")   # fires every day at 07:00 wall-clock time
    """

    def __init__(self, at: str) -> None:
        hour, minute = parse_hh_mm(at, "Daily")
        # 5-field standard cron: minute hour dom month dow
        self._expr = f"{minute} {hour} * * *"
        self._at_str = at
        self._cron = CronTrigger(self._expr)

    def first_run_time(self, current_time: ZonedDateTime) -> ZonedDateTime:
        """Return the next cron-grid-aligned daily run time at or after current_time."""
        return self._cron.first_run_time(current_time)

    def next_run_time(self, previous_run: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime:
        """Return the next daily run time after previous_run that is later than current_time."""
        return self._cron.next_run_time(previous_run, current_time)

    def trigger_label(self) -> str:
        return "daily"

    def trigger_detail(self) -> str | None:
        return self._at_str

    def trigger_db_type(self) -> Literal["cron"]:
        return "cron"

    def trigger_id(self) -> str:
        return f"cron:{self._expr}"

first_run_time(current_time: ZonedDateTime) -> ZonedDateTime

Return the next cron-grid-aligned daily run time at or after current_time.

Source code in src/hassette/scheduler/triggers.py

def first_run_time(self, current_time: ZonedDateTime) -> ZonedDateTime:
    """Return the next cron-grid-aligned daily run time at or after current_time."""
    return self._cron.first_run_time(current_time)

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

Return the next daily run time after previous_run that is later than current_time.

Source code in src/hassette/scheduler/triggers.py

def next_run_time(self, previous_run: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime:
    """Return the next daily run time after previous_run that is later than current_time."""
    return self._cron.next_run_time(previous_run, current_time)

Every

Fixed-interval trigger with drift-resistant scheduling.

Accepts seconds, hours, minutes, or a combination. An optional start parameter anchors the interval grid; if omitted, the first call to first_run_time is used as the anchor.

Parameters:

Name	Type	Description	Default
seconds	float	Interval component in seconds.	0
minutes	float	Interval component in minutes.	0
hours	float	Interval component in hours.	0
start	ZonedDateTime | None	Optional ZonedDateTime anchor for the interval grid. If the anchor is in the past, missed intervals are skipped to produce a near-future run time.	None

Example

Every(hours=1) # every hour, anchored to first run Every(seconds=30, start=my_start_time) # every 30 s, grid anchored to my_start_time

Source code in src/hassette/scheduler/triggers.py

class Every:
    """Fixed-interval trigger with drift-resistant scheduling.

    Accepts seconds, hours, minutes, or a combination. An optional ``start``
    parameter anchors the interval grid; if omitted, the first call to
    ``first_run_time`` is used as the anchor.

    Args:
        seconds: Interval component in seconds.
        minutes: Interval component in minutes.
        hours: Interval component in hours.
        start: Optional ``ZonedDateTime`` anchor for the interval grid. If the
            anchor is in the past, missed intervals are skipped to produce a
            near-future run time.

    Example:
        Every(hours=1)                          # every hour, anchored to first run
        Every(seconds=30, start=my_start_time)  # every 30 s, grid anchored to my_start_time
    """

    def __init__(
        self,
        seconds: float = 0,
        minutes: float = 0,
        hours: float = 0,
        start: ZonedDateTime | None = None,
    ) -> None:
        total = TimeDelta(seconds=seconds, minutes=minutes, hours=hours)
        total_seconds = total.total("seconds")
        if total_seconds <= 0:
            raise ValueError("Every trigger interval must be positive")
        if total_seconds != int(total_seconds):
            raise ValueError("Every trigger interval must be a whole number of seconds")
        self._interval = total
        self._start = start

    @property
    def interval_seconds(self) -> float:
        return self._interval.total("seconds")

    def first_run_time(self, current_time: ZonedDateTime) -> ZonedDateTime:
        """Return the first run time, aligned to the interval grid."""
        start = self._start if self._start is not None else current_time
        if start > current_time:
            return start.round("second")
        return self.advance_past(start, current_time)

    def next_run_time(self, previous_run: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime:
        """Return the next interval tick after previous_run that is later than current_time."""
        return self.advance_past(previous_run, current_time)

    def advance_past(self, anchor: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime:
        """Advance anchor by whole intervals until the result is strictly after current_time."""
        interval_secs = self._interval.total("seconds")
        elapsed = (current_time - anchor).total("seconds")
        if elapsed > 0:
            missed = int(elapsed / interval_secs)
            anchor = anchor.add(seconds=missed * interval_secs)
        result = anchor.add(seconds=interval_secs)
        # Guard: if floating-point truncation landed result at or before current_time,
        # advance one more interval. Boundary-exact slots are treated as "past."
        if result <= current_time:
            result = result.add(seconds=interval_secs)
        return result.round("second")

    def trigger_label(self) -> str:
        return "every"

    def trigger_detail(self) -> str | None:
        return f"{int(self.interval_seconds)}s"

    def trigger_db_type(self) -> Literal["interval"]:
        return "interval"

    def trigger_id(self) -> str:
        return f"every:{int(self.interval_seconds)}"

first_run_time(current_time: ZonedDateTime) -> ZonedDateTime

Return the first run time, aligned to the interval grid.

Source code in src/hassette/scheduler/triggers.py

def first_run_time(self, current_time: ZonedDateTime) -> ZonedDateTime:
    """Return the first run time, aligned to the interval grid."""
    start = self._start if self._start is not None else current_time
    if start > current_time:
        return start.round("second")
    return self.advance_past(start, current_time)

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

Return the next interval tick after previous_run that is later than current_time.

Source code in src/hassette/scheduler/triggers.py

def next_run_time(self, previous_run: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime:
    """Return the next interval tick after previous_run that is later than current_time."""
    return self.advance_past(previous_run, current_time)

advance_past(anchor: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime

Advance anchor by whole intervals until the result is strictly after current_time.

Source code in src/hassette/scheduler/triggers.py

def advance_past(self, anchor: ZonedDateTime, current_time: ZonedDateTime) -> ZonedDateTime:
    """Advance anchor by whole intervals until the result is strictly after current_time."""
    interval_secs = self._interval.total("seconds")
    elapsed = (current_time - anchor).total("seconds")
    if elapsed > 0:
        missed = int(elapsed / interval_secs)
        anchor = anchor.add(seconds=missed * interval_secs)
    result = anchor.add(seconds=interval_secs)
    # Guard: if floating-point truncation landed result at or before current_time,
    # advance one more interval. Boundary-exact slots are treated as "past."
    if result <= current_time:
        result = result.add(seconds=interval_secs)
    return result.round("second")

Once

One-shot trigger that fires at a specific wall-clock time.

Parameters:

Name	Type	Description	Default
at	str | ZonedDateTime	Target time. Accepts a "HH:MM" string (interpreted as today's wall-clock time in the configured timezone — see HassetteConfig.timezone; falls back to the process timezone when unset) or a ZonedDateTime (absolute instant; timezone is explicit).	required
if_past	Literal['tomorrow', 'error']	Behavior when the computed fire time is in the past. - "tomorrow" (default): For "HH:MM" string inputs, defers to the next day. No effect for ZonedDateTime inputs (the absolute instant is used as-is; if it is in the past, the job fires immediately). - "error": Raises ValueError if the computed fire time is in the past. Applies to both "HH:MM" and ZonedDateTime inputs.	'tomorrow'

Example

Once(at="07:00") # fires today at 07:00 (or tomorrow if past) Once(at="07:00", if_past="error") # raises if 07:00 has already passed

Source code in src/hassette/scheduler/triggers.py

class Once:
    """One-shot trigger that fires at a specific wall-clock time.

    Args:
        at: Target time. Accepts a ``"HH:MM"`` string (interpreted as today's
            wall-clock time in the configured timezone — see
            ``HassetteConfig.timezone``; falls back to the process timezone
            when unset) or a ``ZonedDateTime`` (absolute instant; timezone
            is explicit).
        if_past: Behavior when the computed fire time is in the past.
            - ``"tomorrow"`` (default): For ``"HH:MM"`` string inputs, defers to the next day.
              No effect for ``ZonedDateTime`` inputs (the absolute instant is used as-is;
              if it is in the past, the job fires immediately).
            - ``"error"``: Raises ``ValueError`` if the computed fire time is in the past.
              Applies to both ``"HH:MM"`` and ``ZonedDateTime`` inputs.

    Example:
        Once(at="07:00")                      # fires today at 07:00 (or tomorrow if past)
        Once(at="07:00", if_past="error")     # raises if 07:00 has already passed
    """

    def __init__(
        self,
        at: str | ZonedDateTime,
        if_past: Literal["tomorrow", "error"] = "tomorrow",
    ) -> None:
        self._if_past = if_past
        self._at_str: str | None = None

        if isinstance(at, str):
            self._at_str = at
            hour, minute = parse_hh_mm(at, "Once")
            now = date_utils.now()
            target = ZonedDateTime(now.year, now.month, now.day, hour, minute, tz=now.tz)
            if target <= now:
                if if_past == "error":
                    raise ValueError(f"Once(at={at!r}) constructed after the target time and if_past='error'")
                # Defer to tomorrow
                LOGGER.warning(
                    "Once(at=%r) constructed after the target time — deferring to tomorrow.",
                    at,
                )
                target = target.add(days=1)
            self._fire_at = target
        else:
            now = date_utils.now()
            if at <= now:
                if if_past == "error":
                    raise ValueError(f"Once(at=<ZonedDateTime {at.format_iso()!r}>) is in the past and if_past='error'")
                LOGGER.warning(
                    "Once received ZonedDateTime in the past; firing immediately — "
                    "if_past='tomorrow' cannot defer an absolute instant. (at=%r)",
                    at.format_iso(),
                )
            self._fire_at = at

    def first_run_time(self, current_time: ZonedDateTime) -> ZonedDateTime:
        """Return the scheduled fire time."""
        return self._fire_at

    def next_run_time(self, previous_run: ZonedDateTime, current_time: ZonedDateTime) -> None:
        """One-shot trigger; always returns None."""
        return

    def trigger_label(self) -> str:
        return "once"

    def trigger_detail(self) -> str | None:
        if self._at_str is not None:
            return self._at_str
        return self._fire_at.format_iso()

    def trigger_db_type(self) -> Literal["once"]:
        return "once"

    def trigger_id(self) -> str:
        # Always include the full ISO timestamp so two Once jobs constructed on different days
        # (both at "07:00") do not share a trigger_id and do not shadow each other in the heap.
        return f"once:{self._fire_at.format_iso()}"

first_run_time(current_time: ZonedDateTime) -> ZonedDateTime

Return the scheduled fire time.

Source code in src/hassette/scheduler/triggers.py

def first_run_time(self, current_time: ZonedDateTime) -> ZonedDateTime:
    """Return the scheduled fire time."""
    return self._fire_at

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

One-shot trigger; always returns None.

Source code in src/hassette/scheduler/triggers.py

def next_run_time(self, previous_run: ZonedDateTime, current_time: ZonedDateTime) -> None:
    """One-shot trigger; always returns None."""
    return