scheduler

dev_tool.services.scheduler

__all__ = ['SchedulerService'] module-attribute

SchedulerService

Bases: BaseService

A service class for task scheduling operations.

This class provides methods for scheduling, cancelling, and managing time-based tasks with persistence support.

The constructor for the SchedulerService class.

Parameters:

• sqlite (SQLiteDatabase | None, default: None ) –

  The SQLite database for task persistence.

Source code in dev_tool/services/scheduler/service.py

def __init__(self, sqlite: SQLiteDatabase | None = None) -> None:
    """
    The constructor for the SchedulerService class.

    :param sqlite: The SQLite database for task persistence.
    """

    super().__init__()

    self._lock = threading.Lock()
    self._scheduled = {}
    self.sqlite = sqlite

    if self.sqlite:
        try:
            self._restore_schedule()
        except ScheduleRestoreError:
            message = 'Failed to restore scheduled task(s)'
            log.exception(message)

sqlite = sqlite instance-attribute

cancel_task

A method that cancels a scheduled task.

Parameters:

• name (str) –

  The name of the task to cancel.

Returns:

• bool –

  True if the task was cancelled, False if not found.

Raises:

• TaskCancellationError –

  If task cancellation fails.

Source code in dev_tool/services/scheduler/service.py

def cancel_task(self, name: str) -> bool:
    """
    A method that cancels a scheduled task.

    :param name: The name of the task to cancel.
    :return: True if the task was cancelled, False if not found.
    :raises TaskCancellationError: If task cancellation fails.
    """

    try:
        with self._lock:
            if name in self._scheduled:
                self._scheduled[name].cancel()
                del self._scheduled[name]

                message = f'Cancelled task: {name}'
                self.notification.normal_text(message)

                log.debug(message)

                if self.sqlite:
                    self.sqlite.remove_schedule(name)

                return True

            return False
    except Exception:
        message = f'Failed to cancel task: {name}'
        log.exception(message)

        raise TaskCancellationError(message) from None

schedule_task

A method that schedules a task to run after a specified delay.

Parameters:

• name (str) –

  The name of the task.

• task (Callable[[], None]) –

  The function to execute.

• seconds (float) –

  The delay in seconds before execution.

• persist (bool, default: False ) –

  Whether to persist the task across application restarts.

Raises:

• TaskSchedulingError –

  If task scheduling fails.

Source code in dev_tool/services/scheduler/service.py

def schedule_task(self, name: str, task: Callable[[], None], seconds: float, persist: bool = False) -> None:
    """
    A method that schedules a task to run after a specified delay.

    :param name: The name of the task.
    :param task: The function to execute.
    :param seconds: The delay in seconds before execution.
    :param persist: Whether to persist the task across application restarts.
    :raises TaskSchedulingError: If task scheduling fails.
    """

    try:
        with self._lock:
            if name in self._scheduled:
                self._scheduled[name].cancel()

            timer = threading.Timer(
                seconds,
                self._execute_task,
                args=(name, task)
            )

            timer.daemon = True
            self._scheduled[name] = timer
            timer.start()

        if persist and self.sqlite:
            destruction_time = time.time() + seconds
            self.sqlite.add_schedule(name, destruction_time)

        message = f'Scheduled task {name} to run in {seconds} seconds'
        self.notification.normal_text(message)

        log.debug(message)
    except Exception:
        message = f'Failed to schedule task: {name}'
        log.exception(message)

        raise TaskSchedulingError(message) from None

time_remaining

A method that gets the remaining time before a task executes.

Parameters:

• name (str) –

  The name of the task to check.

Returns:

• float | None –

  The remaining time in seconds, or None if not found.

Source code in dev_tool/services/scheduler/service.py

def time_remaining(self, name: str) -> float | None:
    """
    A method that gets the remaining time before a task executes.

    :param name: The name of the task to check.
    :return: The remaining time in seconds, or None if not found.
    """

    if self.sqlite:
        try:
            return self.sqlite.time_remaining(name)
        except Exception:
            message = f'Failed to get time remaining for task: {name}'
            log.exception(message)

            return None

    return 0