Python sched: Schedule Events

Not every scheduled task requires an external service, a distributed queue, or the operating system’s cron facility. For scripts, simulations, tests, and local processes that must call functions at defined times, the Python sched module provides a small standard-library scheduler. It orders events by deadline and priority, waits until the appropriate moment, and invokes callables with arguments supplied by the application.

This guide explains scheduler, enter(), enterabs(), cancellation, blocking and non-blocking execution, and recurring jobs. It complements our articles about Python atexit, tracebacks, contextvars, collections, and slow Python scripts.

What sched provides

A scheduler receives events and decides when each action should run. Everything in sched lives inside the current process. There is no automatic persistence, database, remote worker, or recovery after restart.

The module is suitable for:

• calling a callback after a delay;
• simulating timed events in tests;
• scheduling short work in a local utility;
• organizing timed demonstration steps;
• building a lightweight periodic loop.

Tasks that must survive crashes, distribute load, or run on several machines need persistent scheduling infrastructure.

Create a scheduler

The main class is sched.scheduler.

import sched

scheduler = sched.scheduler()

Current Python versions default to time.monotonic as the clock and time.sleep as the delay function. A monotonic clock prevents civil-time adjustments from moving relative deadlines.

The official sched documentation states that scheduler can be safely used in multi-threaded environments since Python 3.3, although application data shared by callbacks still requires its own synchronization.

Schedule after a delay

enter() accepts a delay, priority, action, positional-argument sequence, and optional keyword arguments.

import sched

scheduler = sched.scheduler()


def notify(message):
    print(message)

scheduler.enter(
    delay=2,
    priority=1,
    action=notify,
    argument=("Two seconds passed",),
)

scheduler.run()

With the default clock and delay function, delays are measured in seconds. The method returns an event object that can later be cancelled.

Positional and keyword arguments

argument should be a sequence, usually a tuple. kwargs receives a dictionary.

def record(name, status="ok"):
    print(name, status)

scheduler.enter(
    1,
    1,
    record,
    argument=("import",),
    kwargs={"status": "complete"},
)

The event stores object references. If a mutable argument changes before execution, the callback observes the changed object.

Schedule at an absolute deadline

enterabs() accepts an absolute value produced by the configured time function.

import time

scheduler = sched.scheduler(
    timefunc=time.monotonic,
    delayfunc=time.sleep,
)

deadline = time.monotonic() + 5
scheduler.enterabs(
    deadline,
    1,
    notify,
    argument=("Monotonic deadline reached",),
)

Do not pass a time.time() timestamp to a scheduler configured with time.monotonic(). The clocks use different origins and purposes.

Monotonic time versus civil time

time.monotonic() only moves forward and is not affected by manual clock changes, NTP corrections, or daylight-saving transitions. It is the right clock for delays and intervals.

time.time() approximates Unix wall-clock time and may be needed for a requirement such as “run at 3:00 PM on this date.” Convert calendar values carefully and account for clock changes.

The official time documentation explains the available clocks. Calendar scheduling with time zones usually needs datetime and zoneinfo before an event reaches sched.

Event priority

When two events have the same deadline, lower priority numbers run first.

scheduler.enter(1, 20, notify, argument=("priority 20",))
scheduler.enter(1, 5, notify, argument=("priority 5",))
scheduler.enter(1, 10, notify, argument=("priority 10",))

Priority does not interrupt an action already running. It only orders events waiting in the queue.

Run the queue

run() processes events in order.

scheduler.run(blocking=True)

In blocking mode, the method waits until all currently queued events have run. Before each event, the delay function receives the remaining wait.

Late events are not dropped

If one action takes longer than the interval to the next event, the scheduler falls behind. Expired events run as soon as possible in their normal order; none are discarded.

import time


def slow_task():
    time.sleep(3)
    print("slow task finished")

scheduler.enter(0, 1, slow_task)
scheduler.enter(1, 1, notify, argument=("late event",))
scheduler.run()

If accumulated delay is unacceptable, send lengthy work to controlled workers or keep callbacks limited to dispatching work.

Non-blocking execution

With blocking=False, run() executes due events and returns the next deadline, or None when the queue is empty.

next_deadline = scheduler.run(blocking=False)

if next_deadline is None:
    print("Queue is empty")
else:
    print("Next deadline:", next_deadline)

This supports integration with another event loop. Interpret the returned value consistently with the Python version and configured time function.

Integrate with a custom loop

import time

while not scheduler.empty():
    deadline = scheduler.run(blocking=False)
    if deadline is None:
        break

    now = time.monotonic()
    delay = max(0, deadline - now)
    process_interface_for(min(delay, 0.1))

A graphical or asynchronous framework usually has its own timers. Schedule the next wake-up through that framework rather than sleeping on its main thread.

Inspect the queue

The queue property returns pending events in execution order.

for event in scheduler.queue:
    print(
        event.time,
        event.priority,
        event.action,
        event.argument,
        event.kwargs,
    )

The returned list is an ordered snapshot. Editing it does not mutate the scheduler’s internal queue.

Check whether the queue is empty

empty() reports whether pending events exist.

if scheduler.empty():
    print("Nothing scheduled")

In a multithreaded application, another thread can change the state immediately after this check.

Cancel an event

Keep the object returned by enter() or enterabs() and pass it to cancel().

event = scheduler.enter(
    30,
    1,
    notify,
    argument=("will not run",),
)

scheduler.cancel(event)

If the event is no longer in the queue, cancel() raises ValueError.

Concurrent cancellation

try:
    scheduler.cancel(event)
except ValueError:
    print("Event already ran or was cancelled")

Another thread may begin the callback between a cancellation decision and the call. Jobs that need a stronger guarantee should check their own cancellation flag before performing side effects.

Recurring work

Sched has no special recurring-job type. A callback schedules its next occurrence.

def run_periodically(interval):
    print("running")
    scheduler.enter(
        interval,
        1,
        run_periodically,
        argument=(interval,),
    )

scheduler.enter(0, 1, run_periodically, argument=(5,))

This measures the interval from the rescheduling moment. A callback that takes two seconds can create a seven-second start-to-start interval.

Avoid drift

Maintain an absolute schedule by calculating the next deadline from the previous one.

def periodic(deadline, interval):
    print("periodic execution")
    following = deadline + interval
    scheduler.enterabs(
        following,
        1,
        periodic,
        argument=(following, interval),
    )

start = time.monotonic() + 1
scheduler.enterabs(start, 1, periodic, argument=(start, 5))

Define how to handle several missed occurrences: run all, skip some, or calculate the next future deadline.

Callback exceptions

If an action raises, run() propagates the exception. Scheduler state remains consistent, and the failed event is not retried automatically.

def protected():
    try:
        operation()
    except Exception:
        logger.exception("Scheduled event failed")

scheduler.enter(1, 1, protected)

Choose propagation or capture according to application responsibility. Critical failures should remain observable.

Using sched with threads

The scheduler queue is thread-safe, so one thread can add events while another calls run(). Callback code and shared objects do not receive automatic protection.

from threading import Lock

lock = Lock()
state = {}


def update(key, value):
    with lock:
        state[key] = value

A blocking run already sleeping may not immediately react to an earlier event inserted by another thread, depending on the surrounding integration. Responsive systems often use short checks or their own wake-up mechanism.

Custom time and delay functions

Custom functions make deterministic testing possible.

class FakeClock:
    def __init__(self):
        self.now = 0.0

    def time(self):
        return self.now

    def sleep(self, seconds):
        self.now += seconds

clock = FakeClock()
scheduler = sched.scheduler(clock.time, clock.sleep)

Tests can advance virtual time instantly instead of waiting in real time.

The delay function receives zero

After every event, scheduler calls the delay function with zero to give other threads an opportunity to run.

A custom delay function must accept zero and should not treat it as an error.

Deterministic tests

events = []

scheduler.enter(5, 1, events.append, argument=("A",))
scheduler.enter(2, 1, events.append, argument=("B",))
scheduler.run()

assert events == ["B", "A"]
assert clock.now == 5

Also test equal deadlines with different priorities, cancellation, exceptions, and events inserted during execution.

Persistence and restart

The queue exists only in memory. All pending events disappear when the process exits.

Jobs that must survive restarts should be stored in a database or durable queue and rebuilt during startup. Avoid serializing arbitrary callables with pickle.

Several machines

Sched does not coordinate distributed locks or prevent duplicate execution in multiple instances. Every process owns an independent queue.

Use a distributed scheduler, leasing table, or task queue when only one instance should perform the work.

Observability

Record at least:

• a logical event identifier;
• planned and actual start time;
• accumulated delay;
• duration;
• status and exception;
• pending queue length.

Do not log arguments containing passwords, tokens, or personal data.

Controlled shutdown

Stop new recurring scheduling, cancel pending events where appropriate, and allow running callbacks to finish.

stopping = False


def recurring(interval):
    run_work()
    if not stopping:
        scheduler.enter(interval, 1, recurring, argument=(interval,))

Coordinate this with signals and the application lifecycle. Atexit can record a final metric but cannot guarantee crash-time execution.

Common mistakes

• Mixing time.time() and time.monotonic().
• Running long callbacks on the scheduler thread.
• Assuming late events will be discarded.
• Cancelling without handling ValueError.
• Creating recurring drift unintentionally.
• Using an in-memory queue for critical jobs.
• Sharing state without locks.
• Treating sched as a distributed scheduler.

Best practices

• Use a monotonic clock for intervals.
• Keep callbacks short and observable.
• Retain event objects for cancellation.
• Define policies for lateness and recurrence.
• Use a fake clock in tests.
• Synchronize shared data.
• Persist jobs that must survive restarts.
• Choose another system for multi-machine work.

Conclusion

The Python sched module provides a simple time-ordered queue for running functions after delays or at absolute deadlines. It orders events by deadline and priority, supports cancellation, exposes pending events, and works with custom clocks.

Its strength is local simplicity. With short callbacks, the correct clock, and explicit lateness, recurrence, and shutdown policies, sched handles internal timers without dependencies. Persistence, distribution, and recovery require a different scheduling system.