Skip to content

Python command scheduler

This mini project implements a small background scheduler for Python commands. It can run a command once, run a command repeatedly, cancel a repeated command by key, and store execution errors in a shared context.

The code lives under scheduler_pkg and is split into small modules:

File Responsibility
commands.py Defines the command interface and the wrapper used for repeated commands.
scheduler_queue.py Stores future commands in time order.
cancellation.py Tracks which keyed commands are still active.
worker.py Runs the scheduler loop in the background thread.
scheduler.py Provides the public CommandScheduler API.
usage_example.py Shows how to create commands and run the scheduler.
exception_handling_example.py Shows how scheduler errors can be captured or reported.

Basic Idea

The scheduler keeps pending commands in a priority queue built with Python's heapq module. When new work is submitted, heapq.heappush inserts a QueuedCommand into the heap. Because QueuedCommand is ordered by run_at first and sequence second, the command with the earliest scheduled time stays at the front of the queue. If two commands have the same run_at value, sequence preserves the order they were added.

The worker can then look at the first heap item to know which command has the highest priority to run next. It does not need to sort the whole queue each time a command is added.

The scheduler is based on the command pattern. A task is represented by a class that inherits from Command and implements execute.

1
2
3
4
5
6
@dataclass
class PrintCommand(Command):
    message: str

    def execute(self, context: SchedulerContext) -> None:
        print(self.message)

The scheduler does not need to know what the command does. It only needs to know when the command should run and whether it should run again later.

There are two main ways to add work:

1
scheduler.submit(PrintCommand("runs once"))

submit runs a command once, after an optional delay.

1
2
3
4
5
scheduler.schedule(
    PrintCommand("runs repeatedly"),
    interval_s=0.5,
    key="printer",
)

schedule wraps the command in a ScheduledCommand, gives it a repeat interval, and requeues it after each successful attempt while it is still active.

Main Classes

Command

Command is the base abstraction for work.

1
2
3
4
5
6
7
class Command(ABC):
    key: ClassVar[str | None] = None
    repeat_interval_s: float | None = None

    @abstractmethod
    def execute(self, scheduler: SchedulerContext) -> Any:
        pass

Every command must implement execute. The scheduler passes a context object to that method, so commands can read or update shared application state.

The optional key is used for cancellation. Commands without a key are always considered active.

ScheduledCommand

ScheduledCommand wraps another command and adds repeat behavior.

1
2
3
4
5
ScheduledCommand(
    command=command,
    repeat_interval_s=interval_s,
    key_override=key,
)

When its execute method runs, it simply delegates to the original command. This keeps the repeat logic outside the command itself.

TimedCommandQueue

TimedCommandQueue stores commands in a heap. A heap is useful here because the worker always needs the command with the earliest run_at time.

Each queued item contains:

  • run_at: when the command should run
  • sequence: a tie breaker for commands scheduled at the same time
  • token: the cancellation token
  • command: the command object

pop_ready only returns commands whose run_at time has passed and whose token is still active.

CancellationRegistry

The cancellation system uses tokens instead of removing old items from the heap.

When a keyed command is submitted, the registry creates a new object token and stores it under that key. When remove(key) is called, the registry replaces the stored token with a new object.

Old queued commands still exist in the heap, but they no longer match the active token. The worker skips them when they reach the front of the queue.

This keeps cancellation simple because the queue does not need to search through all pending items.

SchedulerWorker

SchedulerWorker owns the run loop:

  1. Check whether a command is ready.
  2. If nothing is ready, wait until the next command time or until a wake event.
  3. Execute the command.
  4. Store or report errors.
  5. Requeue repeating commands that are still active.

The worker uses:

  • a Lock to protect shared queue and cancellation data
  • a wake_event so new commands or removals can interrupt waiting
  • a stop_event so stop() can end the loop

CommandScheduler

CommandScheduler is the public API that connects all the lower-level objects.

1
2
3
4
5
6
scheduler = CommandScheduler(context)
scheduler.start()
scheduler.submit(command)
scheduler.schedule(command, interval_s=1.0, key="job")
scheduler.remove("job")
scheduler.stop()

start creates a daemon thread and runs the worker loop inside it. stop sets the stop event, wakes the worker, and joins the thread.

Example Flow

The example in usage_example.py creates an application context and a PrintCommand.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
context = AppContext()
scheduler = CommandScheduler(context)
scheduler.start()

try:
    scheduler.submit(PrintCommand("runs once"))
    scheduler.schedule(PrintCommand("runs repeatedly"), interval_s=0.5, key="printer")
    time.sleep(2.0)
    scheduler.remove("printer")
finally:
    scheduler.stop()

The first command runs once. The second command runs every half second until remove("printer") invalidates its cancellation token. Finally, stop() shuts down the worker thread.

Why The Design Works

The scheduler separates responsibilities:

  • commands define what to do
  • the queue defines when to run
  • the cancellation registry defines whether a keyed command is still valid
  • the worker defines the background execution loop
  • the scheduler class gives users a small public API

This makes the project easy to extend. For example, new command classes can be added without changing the scheduler, and different error handling can be added by passing an on_error callback to CommandScheduler.

Source Code

__init__.py 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from .commands import BasicSchedulerContext, Command, SchedulerContext, ScheduledCommand
from .scheduler_queue import QueuedCommand, TimedCommandQueue
from .scheduler import CommandScheduler
from .worker import ErrorCallback, SchedulerWorker

__all__ = [
    "BasicSchedulerContext",
    "Command",
    "SchedulerContext",
    "CommandScheduler",
    "QueuedCommand",
    "TimedCommandQueue",
    "ErrorCallback",
    "SchedulerWorker",
    "ScheduledCommand",
]
commands.py 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, ClassVar, Protocol


class SchedulerContext(Protocol):
    last_error: BaseException | None





class Command(ABC):
    key: ClassVar[str | None] = None
    repeat_interval_s: float | None = None

    @abstractmethod
    def execute(self, scheduler: SchedulerContext) -> Any:
        pass


@dataclass
class ScheduledCommand(Command):
    command: Command
    repeat_interval_s: float | None = None
    key_override: str | None = None

    @property
    def key(self) -> str | None:
        if self.key_override is not None:
            return self.key_override
        return self.command.key

    def execute(self, scheduler: SchedulerContext) -> Any:
        return self.command.execute(scheduler)
cancellation.py 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from __future__ import annotations

try:
    from .commands import Command
except ImportError:
    from commands import Command


class CancellationRegistry:
    def __init__(self) -> None:
        self._active_tokens: dict[str, object] = {}

    def new_token(self, command: Command) -> object | None:
        if command.key is None:
            return None
        return object()

    def activate(self, command: Command, token: object | None) -> None:
        if command.key is not None and token is not None:
            self._active_tokens[command.key] = token

    def remove(self, key: str) -> None:
        self._active_tokens[key] = object()

    def is_active(self, command: Command, token: object | None) -> bool:
        if command.key is None:
            return True
        return self._active_tokens.get(command.key) is token
scheduler_queue.py 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
from __future__ import annotations

import heapq
import itertools
from collections.abc import Callable
from dataclasses import dataclass, field

try:
    from .commands import Command
except ImportError:
    from commands import Command

@dataclass(order=True)
class QueuedCommand:
    """Command scheduled for execution by ``TimedCommandQueue``.

    Instances are ordered by ``run_at`` and then ``sequence`` so the heap returns
    the earliest command first while preserving insertion order for ties.

    Attributes:
        run_at: Monotonic timestamp when the command becomes ready to run.
        sequence: Insertion-order tiebreaker used only for deterministic heap
            ordering when multiple commands share the same ``run_at`` value.
        token: Optional identity value used by callers to decide whether this
            queued entry is still active when it reaches the front of the heap.
        command: Command object to execute once the entry is ready and active.
    """

    run_at: float
    sequence: int
    token: object | None = field(compare=False)
    command: Command = field(compare=False)


class TimedCommandQueue:
    def __init__(self) -> None:
        self._items: list[QueuedCommand] = []
        self._sequence = itertools.count()

    def push(self, run_at: float, token: object | None, command: Command) -> None:
        item = QueuedCommand(
            run_at=run_at,
            sequence=next(self._sequence),
            token=token,
            command=command,
        )
        heapq.heappush(self._items, item)

    def pop_ready(
        self,
        now: float,
        is_active: Callable[[Command, object | None], bool],
    ) -> QueuedCommand | None:
        while self._items and self._items[0].run_at <= now:
            item = heapq.heappop(self._items)
            if is_active(item.command, item.token):
                return item
        return None

    def next_delay(self, now: float, default_s: float = 0.1) -> float:
        if not self._items:
            return default_s
        return max(0.0, self._items[0].run_at - now)
worker.py 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
from __future__ import annotations

import threading
import time
from collections.abc import Callable

try:
    from .cancellation import CancellationRegistry
    from .commands import Command, SchedulerContext
    from .scheduler_queue import QueuedCommand, TimedCommandQueue
except ImportError:
    from cancellation import CancellationRegistry
    from commands import Command, SchedulerContext
    from scheduler_queue import QueuedCommand, TimedCommandQueue


ErrorCallback = Callable[[BaseException, Command], None]


class SchedulerWorker:
    def __init__(
        self,
        queue: TimedCommandQueue,
        cancellation: CancellationRegistry,
        lock: threading.Lock,
        wake_event: threading.Event,
        stop_event: threading.Event,
        context: SchedulerContext,
        set_last_error: Callable[[BaseException], None],
        on_error: ErrorCallback | None = None,
    ) -> None:
        self._queue = queue
        self._cancellation = cancellation
        self._lock = lock
        self._wake_event = wake_event
        self._stop_event = stop_event
        self._context = context
        self._set_last_error = set_last_error
        self._on_error = on_error

    def run(self) -> None:
        """
        Run the scheduler worker loop, executing commands as they become ready.
        """
        while not self._stop_event.is_set():
            item = self._pop_ready_command()
            if item is None:
                self._wait_until_next_command()
                continue

            try:
                item.command.execute(self._context)
            except BaseException as exc:
                self._handle_error(exc, item.command)

            if item.command.repeat_interval_s is not None:
                if self._cancellation.is_active(item.command, item.token):
                    self._requeue(item.command, item.token)

    def _pop_ready_command(self) -> QueuedCommand | None:
        now = time.monotonic()
        with self._lock:
            return self._queue.pop_ready(now, self._cancellation.is_active)

    def _wait_until_next_command(self) -> None:
        with self._lock:
            delay_s = self._queue.next_delay(time.monotonic())

        self._wake_event.wait(delay_s)
        self._wake_event.clear()

    def _requeue(self, command: Command, token: object | None) -> None:
        if command.repeat_interval_s is None:
            return

        run_at = time.monotonic() + command.repeat_interval_s
        with self._lock:
            self._queue.push(run_at, token, command)
        self._wake_event.set()

    def _handle_error(self, exc: BaseException, command: Command) -> None:
        if self._on_error is not None:
            self._on_error(exc, command)
            return
        self._set_last_error(exc)
scheduler.py 
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
from __future__ import annotations

import math
import threading
import time

from loguru import logger

try:
    from .cancellation import CancellationRegistry
    from .commands import  Command, ScheduledCommand, SchedulerContext
    from .scheduler_queue import TimedCommandQueue
    from .worker import ErrorCallback, SchedulerWorker
except ImportError:
    from cancellation import CancellationRegistry
    from commands import  Command, ScheduledCommand, SchedulerContext
    from scheduler_queue import TimedCommandQueue
    from worker import ErrorCallback, SchedulerWorker


def _validate_delay(delay_s: float) -> None:
    if not math.isfinite(delay_s) or delay_s < 0:
        raise ValueError("delay_s must be a finite number >= 0")


def _validate_interval(interval_s: float) -> None:
    if not math.isfinite(interval_s) or interval_s <= 0:
        raise ValueError("interval_s must be a finite number > 0")

class CommandScheduler:
    def __init__(
        self,
        context: SchedulerContext | None = None,
        on_error: ErrorCallback | None = None,
        name: str = "scheduler",
    ) -> None:
        self.name = name
        self.context = context
        self._queue = TimedCommandQueue()
        self._cancellation = CancellationRegistry()
        self._lock = threading.Lock()
        self._wake_event = threading.Event()
        self._stop_event = threading.Event()
        self._worker = SchedulerWorker(
            queue=self._queue,
            cancellation=self._cancellation,
            lock=self._lock,
            wake_event=self._wake_event,
            stop_event=self._stop_event,
            context=self.context,
            set_last_error=self._set_last_error,
            on_error=on_error,
        )
        self._thread: threading.Thread | None = None

    # region Public API
    def start(self) -> None:
        if self._thread is not None and self._thread.is_alive():
            logger.info("scheduler already started name={}", self.name)
            return

        self._stop_event.clear()
        self._thread = threading.Thread(
            target=self._worker.run,
            daemon=True,
            name=f"{self.name}_t",
        )
        self._thread.start()
        logger.info("scheduler started name={}", self.name)

    def stop(self, timeout: float | None = 2.0) -> None:
        logger.info("scheduler stopping name={}", self.name)
        self._stop_event.set()
        self._wake_event.set()
        if self._thread is not None:
            self._thread.join(timeout=timeout)
            self._thread = None
        logger.info("scheduler stopped name={}", self.name)

    def submit(self, command: Command, delay_s: float = 0.0) -> None:
        """
        submit a command to be executed after a delay or immediately.
        """
        _validate_delay(delay_s)
        token = self._cancellation.new_token(command)
        self._submit(command, delay_s=delay_s, token=token, replace=True)

    def schedule(
        self,
        command: Command,
        interval_s: float,
        delay_s: float = 0.0,
        key: str | None = None,
    ) -> None:
        """
        schedule a command to be executed repeatedly at a given interval.
        """
        _validate_interval(interval_s)
        _validate_delay(delay_s)

        self.submit(
            ScheduledCommand(
                command=command,
                repeat_interval_s=interval_s,
                key_override=key,
            ),
            delay_s=delay_s,
        )

    def remove(self, key: str) -> None:
        """
        remove a scheduled command by its key.
        """
        with self._lock:
            self._cancellation.remove(key)
        logger.debug("scheduler canceled command name={} key={}", self.name, key)
        self._wake_event.set()
    # endregion

    def _submit(
        self,
        command: Command,
        delay_s: float,
        token: object | None,
        replace: bool,
    ) -> None:
        """
        submit command to schduler queue
        """
        run_at = time.monotonic() + delay_s
        with self._lock:
            if replace:
                self._cancellation.activate(command, token)
            self._queue.push(run_at, token, command)
        logger.debug(
            "scheduler added command name={} command={} delay_s={} run_at={} repeat_interval_s={}",
            self.name,
            command.__class__.__name__,
            delay_s,
            run_at,
            command.repeat_interval_s,
        )
        self._wake_event.set()

    def _set_last_error(self, exc: BaseException) -> None:
        self.context.last_error = exc

Entry Point

usage_example.py 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
from __future__ import annotations

import sys
import time
from dataclasses import dataclass, field
from typing import cast

from loguru import logger

logger.remove()
logger.add(sys.stderr, level="DEBUG")


try:
    from . import  Command, CommandScheduler, SchedulerContext
except ImportError:
    from commands import  Command, SchedulerContext
    from scheduler import CommandScheduler

@dataclass
class BasicSchedulerContext:
    last_error: BaseException | None = None

@dataclass
class AppContext(BasicSchedulerContext):
    messages: list[str] = field(default_factory=list)


@dataclass
class PrintCommand(Command):
    message: str

    def execute(self, context: SchedulerContext) -> None:
        app_context = cast(AppContext, context)
        app_context.messages.append(self.message)
        logger.info(self.message)


REPEAT_CMD_KEY = "printer"

def main() -> None:
    context = AppContext()
    scheduler = CommandScheduler(name="my-scheduler", context=context)
    scheduler.start()

    try:
        cmd_once = PrintCommand(message="runs once")
        cmd_repeated = PrintCommand(message="runs repeatedly")
        scheduler.submit(cmd_once, delay_s=1.0)
        scheduler.schedule(cmd_repeated, interval_s=0.5, key=REPEAT_CMD_KEY)
        time.sleep(2.0)
        scheduler.remove(REPEAT_CMD_KEY)
    finally:
        scheduler.stop()

    print(f"messages stored in context: {len(context.messages)}")


if __name__ == "__main__":
    main()
exception_handling_example.py 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
from __future__ import annotations

import sys
import time
from dataclasses import dataclass, field
from typing import cast

from loguru import logger

try:
    from .commands import Command, SchedulerContext
    from .scheduler import CommandScheduler
except ImportError:
    from commands import Command, SchedulerContext
    from scheduler import CommandScheduler


logger.remove()
logger.add(sys.stderr, level="DEBUG")


@dataclass
class AppContext:
    last_error: BaseException | None = None
    errors: list[str] = field(default_factory=list)
    runs: int = 0


@dataclass
class FailsEverySecondCallCommand(Command):
    job_name: str

    def execute(self, context: SchedulerContext) -> None:
        app_context = cast(AppContext, context)
        app_context.runs += 1
        logger.info("{} run #{}", self.job_name, app_context.runs)

        if app_context.runs % 2 == 0:
            app_context.errors.append(f"{self.job_name} failed on run #{app_context.runs}")
            raise RuntimeError(f"{self.job_name} failed on run #{app_context.runs}")


def handle_scheduler_error(exc: BaseException, command: Command) -> None:
    logger.error("command {} failed: {}", command.__class__.__name__, exc)


def main() -> None:
    command_key = "nightly-report"
    context = AppContext()
    scheduler = CommandScheduler(context, on_error=handle_scheduler_error)
    scheduler.start()

    try:
        scheduler.schedule(
            FailsEverySecondCallCommand(job_name="nightly-report"),
            interval_s=1.0,
            key=command_key,
        )
        time.sleep(5.0)
        scheduler.remove(command_key)

    finally:
        scheduler.stop()
        logger.info("scheduler stopped")

    logger.info("total command runs: {}", context.runs)
    logger.info("errors recorded in context: {}", len(context.errors))


if __name__ == "__main__":
    main()