API Reference¶

def start(self, paused: bool = False) -> bool:
    if not self.event_loop:
        try:
            if self.isolated_event_loop:
                raise RuntimeError()
            self.event_loop = asyncio.get_running_loop()
        except RuntimeError:
            event = Event()
            self.event_loop_thread = Thread(
                target=self._init_new_loop, args=[event], daemon=True
            )
            self.event_loop_thread.start()
            event.wait()
    return super().start(paused)

def shutdown(self, wait: bool = True) -> bool:
    # not decremented yet so +1
    result = self.ref_counter
    self._shutdown(wait)
    return result > 1

def start_timer(self, wait_seconds: Optional[float] = None) -> None:
    self.stop_timer()
    if wait_seconds is not None:
        self.timer = self.event_loop.call_later(wait_seconds, self.wakeup)

def stop_timer(self) -> None:
    if getattr(self, "timer", None):
        self.timer.cancel()  # type: ignore
        self.timer = None

@run_in_event_loop
def wakeup(self) -> None:
    self.stop_timer()
    wait_seconds = self.process_tasks()
    self.start_timer(wait_seconds)

def create_default_executor(self) -> BaseExecutor:
    return AsyncIOExecutor()

def pause(self) -> None:
    """
    Pause task processing in the scheduler.

    This will prevent the scheduler from waking up to do task processing until resume
    is called. It will not however stop any already running task processing.
    """
    if self.state == SchedulerState.STATE_STOPPED:
        raise SchedulerNotRunningError()
    elif self.state == SchedulerState.STATE_RUNNING:
        self.state = SchedulerState.STATE_PAUSED
        self.loggers[self.logger_name].info("Paused scheduler task processing.")
        self.dispatch_event(SchedulerEvent(code=SCHEDULER_PAUSED))

def resume(self) -> None:
    """
    Resume task processing in the scheduler.
    """
    if self.state == SchedulerState.STATE_STOPPED:
        raise SchedulerNotRunningError
    elif self.state == SchedulerState.STATE_PAUSED:
        self.state = SchedulerState.STATE_RUNNING
        self.loggers[self.logger_name].info("Resumed scheduler task processing.")
        self.dispatch_event(SchedulerEvent(code=SCHEDULER_RESUMED))
        self.wakeup()

def add_executor(
    self,
    executor: Union[ExecutorType, str],
    alias: str = "default",
    **executor_options: Any,
) -> None:
    with self.executor_lock:
        if alias in self.executors:
            raise ValueError(
                f"This scheduler already has an executor by the alias of '{alias}'."
            )

        if isinstance(executor, ExecutorType):
            self.executors[alias] = executor
        elif isinstance(executor, str):
            self.executors[alias] = executor = self.create_plugin_instance(
                PluginInstance.EXECUTOR, executor, executor_options
            )
        else:
            raise TypeError(
                f"Expected an executor instance or a string, got {executor.__class__.__name__} instead."
            )
        if self.state != SchedulerState.STATE_STOPPED:
            cast("ExecutorType", executor).start(self, alias)

    self.dispatch_event(SchedulerEvent(code=EXECUTOR_ADDED, alias=alias))

def remove_executor(self, alias: str, shutdown: bool = True) -> None:
    """
    Removes the executor by the given alias from this scheduler.
    """
    with self.executor_lock:
        executor = self.lookup_executor(alias)
        del self.executors[alias]

    if shutdown:
        executor.shutdown()

    self.dispatch_event(SchedulerEvent(code=EXECUTOR_REMOVED, alias=alias))

def add_store(
    self, store: Union[StoreType, str], alias: str = "default", **store_options: Any
) -> None:
    """
    Adds a task store to this scheduler.

    Any extra keyword arguments will be passed to the task store plugin's constructor, assuming
    that the first argument is the name of a task store plugin.
    """
    with self.store_lock:
        if alias in self.stores:
            raise ValueError(
                f"This scheduler already has a task store by the alias of '{alias}'."
            )

        if isinstance(store, StoreType):
            self.stores[alias] = store
        elif isinstance(store, str):
            self.stores[alias] = store = self.create_plugin_instance(
                PluginInstance.STORE, store, store_options
            )
        else:
            raise TypeError(
                f"Expected a task store instance or a string, got {store.__class__.__name__} instead."
            )

        if self.state != SchedulerState.STATE_STOPPED:
            cast("StoreType", store).start(self, alias)

    self.dispatch_event(SchedulerEvent(code=STORE_ADDED, alias=alias))

    if self.state == SchedulerState.STATE_RUNNING:
        self.wakeup()

def remove_store(self, alias: str, shutdown: bool = True) -> None:
    """
    Removes the task store by the given alias from this scheduler.
    """
    with self.store_lock:
        store = self.lookup_store(alias)
        del self.stores[alias]

    if shutdown:
        store.shutdown()

    self.dispatch_event(SchedulerEvent(code=STORE_REMOVED, alias=alias))

def add_listener(self, callback: Any, mask: Union[int, str] = ALL_EVENTS) -> None:
    """
    add_listener(callback, mask=EVENT_ALL)

    Adds a listener for scheduler events.

    When a matching event  occurs, callback is executed with the event object as its
    sole argument. If the mask parameter is not provided, the callback will receive events
    of all types.

    Args:
        callback: any callable that takes one argument.
        mask: bitmask that indicates which events should be listened to.
    """
    with self.listeners_lock:
        self.listeners.append((callback, mask))

def remove_listener(self, callback: Any) -> bool:
    """
    Removes a previously added event listener.
    """
    with self.listeners_lock:
        for index, (_callback, _) in enumerate(self.listeners):
            if callback == _callback:
                del self.listeners[index]
                return True
    return False

def add_task(
    self,
    fn_or_task: Optional[Union[Callable[..., Any], TaskType]] = None,
    trigger: Optional[Union[TriggerType, str]] = None,
    args: Optional[Any] = None,
    kwargs: Optional[Any] = None,
    id: Optional[str] = None,
    name: Optional[str] = None,
    mistrigger_grace_time: Union[int, Undefined, None] = undefined,
    coalesce: Union[bool, Undefined] = undefined,
    max_instances: Union[int, Undefined, None] = undefined,
    next_run_time: Union[datetime, str, Undefined, None] = undefined,
    store: Union[str, Undefined, None] = None,
    executor: Union[str, Undefined, None] = None,
    replace_existing: bool = False,
    # old name
    fn: Optional[Any] = None,
    **trigger_args: Any,
) -> TaskType:
    """
    Adds the given task to the task list and wakes up the scheduler if it's already running.

    Any option that defaults to undefined will be replaced with the corresponding default
    value when the task is scheduled (which happens when the scheduler is started, or
    immediately if the scheduler is already running).

    The fn argument can be given either as a callable object or a textual reference in
    the package.module:some.object format, where the first half (separated by :) is an
    importable module and the second half is a reference to the callable object, relative to
    the module.

    The trigger argument can either be:
      . The alias name of the trigger (e.g. date, interval or cron), in which case
        any extra keyword arguments to this method are passed on to the trigger's constructor.
      . An instance of a trigger class (TriggerType).


    Args:
        fn: Callable (or a textual reference to one) to run at the given time.
        trigger: Trigger instance that determines when fn is called.
        args: List of positional arguments to call fn with.
        kwargs: Dict of keyword arguments to call fn with.
        id: Explicit identifier for the task (for modifying it later).
        name: Textual description of the task.
        mistriger_grace_time: Seconds after the designated runtime that the task is still
            allowed to be run (or None to allow the task to run no matter how late it is).
        coalesce: Run once instead of many times if the scheduler determines that the
            task should be run more than once in succession.
        max_instances: Maximum number of concurrently running instances allowed for this task.
        next_run_time: When to first run the task, regardless of the trigger (pass
            None to add the task as paused).
        store: Alias of the task store to store the task in.
        executor: Alias of the executor to run the task with.
        replace_existing: True to replace an existing task with the same id
                          (but retain the number of runs from the existing one).
    """
    from asyncz.tasks import Task

    if fn is not None:
        warnings.warn(
            "The parameter 'fn' was renamed to 'fn_or_task'",
            DeprecationWarning,
            stacklevel=2,
        )
        fn_or_task = fn
    if isinstance(fn_or_task, TaskType):
        assert fn_or_task.submitted is False, "Can submit tasks only once"
        fn_or_task.submitted = True
        # tweak task before submitting
        # WARNING: in contrast to the decorator mode this really updates the task
        # by providing an id (e.g. autogenerated) you can make a real task from decorator type task while submitting
        task_update_kwargs: dict[str, Any] = {
            "scheduler": self,
            "args": tuple(args) if args is not None else undefined,
            "kwargs": dict(kwargs) if kwargs is not None else undefined,
            "id": id or undefined,
            "name": name or undefined,
            "mistrigger_grace_time": mistrigger_grace_time,
            "coalesce": coalesce,
            "max_instances": max_instances,
            "next_run_time": next_run_time,
            "executor": executor if executor is not None else undefined,
            "store_alias": store if store is not None else undefined,
        }
        if trigger:
            task_update_kwargs["trigger"] = self.create_trigger(trigger, trigger_args)
            # WARNING: when submitting a task object allow_mistrigger_by_default has no effect
        task_update_kwargs = {
            key: value for key, value in task_update_kwargs.items() if value is not undefined
        }
        fn_or_task.update_task(**task_update_kwargs)
        # fallback if still not set. Set manually executor and store_alias to default.
        if fn_or_task.executor is None:
            fn_or_task.executor = "default"
        if fn_or_task.store_alias is None:
            fn_or_task.store_alias = "default"
        assert fn_or_task.trigger is not None, "Cannot submit a task without a trigger."
        assert fn_or_task.id is not None, "Cannot submit a decorator type task."
        with self.store_lock:
            if self.state == SchedulerState.STATE_STOPPED:
                self.pending_tasks.append(
                    (
                        fn_or_task,
                        replace_existing,
                        next_run_time is not None,
                    )
                )
                self.loggers[self.logger_name].info(
                    "Adding task tentatively. It will be properly scheduled when the scheduler starts."
                )
            else:
                self.real_add_task(
                    fn_or_task, replace_existing, next_run_time is not None, None
                )
        return fn_or_task
    task_kwargs: dict[str, Any] = {
        "scheduler": self,
        "trigger": self.create_trigger(trigger, trigger_args),
        "fn": fn_or_task,
        "args": tuple(args) if args is not None else (),
        "kwargs": dict(kwargs) if kwargs is not None else {},
        "id": id,
        "name": name,
        "mistrigger_grace_time": mistrigger_grace_time,
        "coalesce": coalesce,
        "max_instances": max_instances,
        "next_run_time": next_run_time,
        "executor": executor if executor is not None else undefined,
        "store_alias": store if store is not None else undefined,
    }
    task_kwargs = {key: value for key, value in task_kwargs.items() if value is not undefined}
    if task_kwargs["trigger"].allow_mistrigger_by_default:
        # we want to be able, to just use add_task, and the task is scheduled
        task_kwargs.setdefault("mistrigger_grace_time", None)
    for key, value in self.task_defaults.model_dump(exclude_none=True).items():
        task_kwargs.setdefault(key, value)

    task = Task(**task_kwargs)
    if task.fn is not None:
        return self.add_task(
            task, replace_existing=replace_existing, next_run_time=next_run_time
        )
    return task

def update_task(
    self, task_id: Union[TaskType, str], store: Optional[str] = None, **updates: Any
) -> TaskType:
    """
    Modifies the properties of a single task.

    Modifications are passed to this method as extra keyword arguments.

    Args:
        task_id: The identifier of the task.
        store: Alias of the store that contains the task.
    """
    if isinstance(task_id, TaskType):
        assert task_id.id, "Cannot update a decorator type Task"
        new_updates = task_id.model_dump()
        new_updates.update(**updates)
        task_id = task_id.id
    else:
        new_updates = updates

    with self.store_lock:
        task, store = self.lookup_task(task_id, store)
        task.update_task(scheduler=self, **new_updates)

        if store:
            self.lookup_store(store).update_task(task)

    self.dispatch_event(TaskEvent(code=TASK_MODIFIED, task_id=task_id, store=store))

    if self.state == SchedulerState.STATE_RUNNING:
        self.wakeup()
    return task

def reschedule_task(
    self,
    task_id: Union[TaskType, str],
    store: Optional[str] = None,
    trigger: Optional[Union[str, TriggerType]] = None,
    **trigger_args: Any,
) -> TaskType:
    """
    Constructs a new trigger for a task and updates its next run time.

    Extra keyword arguments are passed directly to the trigger's constructor.

    Args:
        task_id: The identifier of the task.
        store: Alias of the task store that contains the task.
        trigger: Alias of the trigger type or a trigger instance.
    """
    trigger = self.create_trigger(trigger, trigger_args)
    now = datetime.now(self.timezone)
    next_run_time = trigger.get_next_trigger_time(self.timezone, None, now)
    return self.update_task(task_id, store, trigger=trigger, next_run_time=next_run_time)

def pause_task(self, task_id: Union[TaskType, str], store: Optional[str] = None) -> TaskType:
    """
    Causes the given task not to be executed until it is explicitly resumed.

    Args:
        task_id: The identifier of the task.
        store: Alias of the task store that contains the task.
    """
    return self.update_task(task_id, store, next_run_time=None)

def resume_task(
    self, task_id: Union[TaskType, str], store: Optional[str] = None
) -> Union[TaskType, None]:
    """
    Resumes the schedule of the given task, or removes the task if its schedule is finished.

    Args:
        task_id: The identifier of the task.
        store: Alias of the task store that contains the task.
    """
    if isinstance(task_id, TaskType):
        assert task_id.id, "Cannot resume decorator style Task"
        task_id = task_id.id
    with self.store_lock:
        task, store = self.lookup_task(task_id, store)
        now = datetime.now(self.timezone)
        next_run_time = task.trigger.get_next_trigger_time(self.timezone, None, now)  # type: ignore

        if next_run_time:
            return self.update_task(task_id, store, next_run_time=next_run_time)
        else:
            self.delete_task(task.id, store)
            return None

def run_task(
    self,
    task_id: Union[TaskType, str],
    store: Optional[str] = None,
    *,
    force: bool = True,
    remove_finished: bool = True,
) -> Union[TaskType, None]:
    """
    Submit a task to its configured executor immediately.

    This method centralizes the "run now" behavior that operational surfaces
    such as the CLI and dashboard need. It reuses the task's configured
    executor, dispatches the same submission-related events that the main
    scheduler loop emits, and persists the task's updated schedule state.

    Args:
        task_id: The task instance or identifier to execute immediately.
        store: Optional preferred store alias when resolving the task.
        force: When ``True`` and the trigger has no run times due yet, submit
            a one-off immediate execution at ``now``.
        remove_finished: When ``True`` and the trigger yields no further run
            times after this execution, delete the task. When ``False``,
            preserve the task in a paused state.

    Returns:
        The updated task when it remains scheduled or paused, or ``None`` if
        it was removed because the schedule finished and ``remove_finished``
        was requested.

    Raises:
        SchedulerNotRunningError: If the scheduler has not been started yet.
        MaximumInstancesError: If the task is already at its executor's
            concurrency limit.
        TaskLookupError: If the task cannot be found.
        KeyError: If the configured executor cannot be found.
    """

    if self.state == SchedulerState.STATE_STOPPED:
        raise SchedulerNotRunningError()

    if isinstance(task_id, TaskType):
        assert task_id.id, "Cannot run a decorator style Task."
        task_id = task_id.id

    task, store_alias = self.lookup_task(task_id, store)
    now = datetime.now(self.timezone)
    run_times = task.get_run_times(self.timezone, now)
    if not run_times:
        if not force:
            return task
        run_times = [now]

    assert task.executor is not None, "Task has no executor configured."
    executor = self.lookup_executor(task.executor)

    try:
        executor.send_task(task, run_times)
    except MaximumInstancesError:
        self.loggers[self.logger_name].warning(
            "Execution of task '%s' skipped: maximum running instances reached (%s).",
            task,
            task.max_instances,
        )
        self.dispatch_event(
            TaskSubmissionEvent(
                code=TASK_MAX_INSTANCES,
                task_id=task_id,
                store=store_alias,
                scheduled_run_times=run_times,
            )
        )
        raise
    except Exception:
        self.loggers[self.logger_name].exception(
            "Error submitting task '%s' to executor '%s'.",
            task,
            task.executor,
        )
        raise
    else:
        self.dispatch_event(
            TaskSubmissionEvent(
                code=TASK_SUBMITTED,
                task_id=task_id,
                store=store_alias,
                scheduled_run_times=run_times,
            )
        )

    next_run_time = task.trigger.get_next_trigger_time(  # type: ignore[union-attr]
        self.timezone,
        run_times[-1],
        now,
    )

    if next_run_time is not None and next_run_time <= now:
        # Manual runs must always advance strictly forward so admin actions do
        # not create zero-delay feedback loops in tests or dashboards.
        next_run_time = now + timedelta(milliseconds=1)

    if next_run_time is None and remove_finished:
        self.delete_task(task_id, store_alias)
        return None

    task.update_task(next_run_time=next_run_time)
    if store_alias is not None:
        with self.store_lock:
            self.lookup_store(store_alias).update_task(task)

    return task

def get_tasks(self, store: Optional[str] = None) -> list[TaskType]:
    """
    Returns a list of pending tasks (if the scheduler hasn't been started yet) and scheduled
    tasks, either from a specific task store or from all of them.

    If the scheduler has not been started yet, only pending tasks can be returned because the
    task stores haven't been started yet either.

    Args:
        store: alias of the task store.
    """
    with self.store_lock:
        tasks = []
        if self.state == SchedulerState.STATE_STOPPED:
            for task, _, _ in self.pending_tasks:
                if store is None or task.store_alias == store:
                    tasks.append(task)
        else:
            for alias, _store in self.stores.items():
                if store is None or alias == store:
                    tasks.extend(_store.get_all_tasks())

        return tasks

def get_task(self, task_id: str, store: Optional[str] = None) -> Union[TaskType, None]:
    """
    Returms the Task that matches the given task_id.

    Args:
        task_id: The identifier of the task.
        store: Alias of the task store that most likely contains the task.
    """
    with self.store_lock:
        try:
            return self.lookup_task(task_id, store)[0]
        except TaskLookupError:
            return None

def get_task_info(self, task_id: str, store: Optional[str] = None) -> Union[TaskInfo, None]:
    """
    Return an immutable inspection snapshot for a single task.

    The returned snapshot is safe to expose in operational tooling because it
    is detached from the underlying task object and captures only observable
    scheduler metadata.
    """

    task = self.get_task(task_id, store)
    return task.snapshot() if task is not None else None

def get_task_infos(
    self,
    store: Optional[str] = None,
    *,
    schedule_state: TaskScheduleState | str | None = None,
    executor: str | None = None,
    trigger: str | None = None,
    q: str | None = None,
    sort_by: str = "next_run_time",
    descending: bool = False,
) -> list[TaskInfo]:
    """
    Return task snapshots with scheduler-native filtering and sorting.

    This API is meant for read-oriented operational tooling. Instead of every
    consumer re-implementing its own serialization, filtering, and ordering
    logic, the scheduler exposes a single consistent view of task metadata.

    Args:
        store: Restrict results to one store alias.
        schedule_state: Optional state filter. Accepts a
            ``TaskScheduleState`` member or its string value.
        executor: Restrict results to one executor alias.
        trigger: Restrict results to one trigger alias or trigger class name.
        q: Case-insensitive free-text search across task identifiers, names,
            callable information, trigger metadata, executor, and state.
        sort_by: Field to sort on. Supported values are ``id``, ``name``,
            ``next_run_time``, ``schedule_state``, ``executor``, ``store``,
            and ``trigger``.
        descending: Reverse the final sort order.
    """

    resolved_state: TaskScheduleState | None = None
    if schedule_state is not None:
        resolved_state = (
            schedule_state
            if isinstance(schedule_state, TaskScheduleState)
            else TaskScheduleState(str(schedule_state).strip().lower())
        )

    infos = [task.snapshot() for task in self.get_tasks(store)]
    if resolved_state is not None:
        infos = [info for info in infos if info.schedule_state is resolved_state]

    if executor:
        infos = [info for info in infos if info.executor == executor]

    if trigger:
        trigger_value = trigger.strip().lower()
        infos = [
            info
            for info in infos
            if (info.trigger_alias or "").lower() == trigger_value
            or (info.trigger_name or "").lower() == trigger_value
        ]

    if q:
        needle = q.strip().lower()
        infos = [
            info
            for info in infos
            if needle in (info.id or "").lower()
            or needle in (info.name or "").lower()
            or needle in (info.callable_name or "").lower()
            or needle in (info.callable_reference or "").lower()
            or needle in (info.trigger_alias or "").lower()
            or needle in (info.trigger_name or "").lower()
            or needle in (info.trigger_description or "").lower()
            or needle in (info.executor or "").lower()
            or needle in (info.store_alias or "").lower()
            or needle in info.schedule_state.value
        ]

    sorters: dict[str, Callable[[TaskInfo], Any]] = {
        "id": lambda info: (info.id or "").lower(),
        "name": lambda info: ((info.name or "").lower(), (info.id or "").lower()),
        "next_run_time": lambda info: (
            info.next_run_time is None,
            info.next_run_time or datetime.max.replace(tzinfo=self.timezone),
            (info.name or "").lower(),
            (info.id or "").lower(),
        ),
        "schedule_state": lambda info: (
            info.schedule_state.value,
            info.next_run_time is None,
            info.next_run_time or datetime.max.replace(tzinfo=self.timezone),
            (info.id or "").lower(),
        ),
        "executor": lambda info: ((info.executor or "").lower(), (info.id or "").lower()),
        "store": lambda info: ((info.store_alias or "").lower(), (info.id or "").lower()),
        "trigger": lambda info: (
            (info.trigger_alias or info.trigger_name or "").lower(),
            (info.id or "").lower(),
        ),
    }
    try:
        key = sorters[sort_by]
    except KeyError as exc:
        raise ValueError(
            "sort_by must be one of: id, name, next_run_time, schedule_state, "
            "executor, store, trigger."
        ) from exc

    return sorted(infos, key=key, reverse=descending)