Scheduler

raw JSON →
0.8.11 verified Sat May 09 auth: no python

A simple in-process Python scheduler library with asyncio, threading, and timezone support. Current version 0.8.11 supports Python >=3.10. Released as needed.

pip install scheduler
error AttributeError: module 'scheduler' has no attribute 'Scheduler'
cause Importing the wrong path (e.g., from scheduler.scheduler import Scheduler) in older versions or using a stale installation.
fix
Use 'from scheduler import Scheduler' (top-level).
error AttributeError: 'Scheduler' object has no attribute 'schedule'
cause Calling the old method name 'schedule' which was renamed to 'cyclic' in v0.8.0.
fix
Replace 'scheduler.schedule(interval, job)' with 'scheduler.cyclic(interval, job)'.
error ImportError: cannot import name 'AsyncIOScheduler' from 'scheduler'
cause Trying to import the async scheduler directly from the top-level package, but it is in a submodule.
fix
Use 'from scheduler.asyncio import Scheduler as AsyncIOScheduler'.
gotcha The library requires manual calling of exec_jobs() in a loop; it does not run automatically. Failing to call exec_jobs() in a loop will result in no tasks executing.
fix Run scheduler.exec_jobs() inside a while loop or use the provided async/threading wrappers.
gotcha Timing changes between exec_jobs() calls are skipped if the next execution time is in the past. Tasks are not caught up; they wait for the next scheduled time.
fix Ensure exec_jobs() is called frequently enough to avoid missing scheduled times, or adjust for missed executions manually.
deprecated The method 'schedule' was renamed to 'cyclic' in version 0.8.0. Using 'schedule' raises AttributeError.
fix Use scheduler.cyclic(interval, job) instead of scheduler.schedule(interval, job).
breaking Support for Python <3.10 dropped in version 0.8.0. Installing on older Python versions will fail.
fix Upgrade to Python >=3.10 or pin to scheduler==0.7.3.

Basic usage: create a Scheduler, schedule a cyclic task, then repeatedly call exec_jobs() in a loop.

from scheduler import Scheduler
import time

scheduler = Scheduler()

def my_task():
    print("Task executed")

# Schedule every 5 seconds
scheduler.cyclic(5, my_task)

# Run for 10 seconds
start = time.time()
while time.time() - start < 10:
    scheduler.exec_jobs()
    time.sleep(0.1)