philiprehberger-task-scheduler

Cron-like task scheduler with overlap prevention and interval support.

Installation

pip install philiprehberger-task-scheduler

Usage

from philiprehberger_task_scheduler import Scheduler

scheduler = Scheduler()

@scheduler.cron("*/5 * * * *")  # every 5 minutes
def check_health():
    ping_server()

scheduler.start()  # blocks

Interval Scheduling

from philiprehberger_task_scheduler import Scheduler

scheduler = Scheduler()

@scheduler.interval(seconds=30)
def poll_queue():
    process_messages()

@scheduler.interval(minutes=5, overlap=False)
def sync_data():
    pull_latest_data()

One-Shot Tasks

from philiprehberger_task_scheduler import Scheduler

scheduler = Scheduler()

@scheduler.once(delay=10)  # run once after 10 seconds
def startup_task():
    warm_cache()

Background Mode

from philiprehberger_task_scheduler import Scheduler

scheduler = Scheduler()
scheduler.start(background=True)
# ... your app continues running ...
scheduler.stop()

Programmatic API

from philiprehberger_task_scheduler import Scheduler

scheduler = Scheduler()
scheduler.add("my-job", fn=my_function, cron="0 * * * *")
scheduler.add("poller", fn=poll, interval_seconds=60)
scheduler.remove("my-job")

Execution History

from philiprehberger_task_scheduler import Scheduler

scheduler = Scheduler(history_limit=50)
scheduler.add("job", fn=my_function, interval_seconds=60)
scheduler.start(background=True)

# Get all history (newest first)
for record in scheduler.history:
    print(f"{record.job_name}: {record.status.value} in {record.duration_seconds:.2f}s")

# Get history for a specific job
for record in scheduler.get_job_history("job"):
    if record.error:
        print(f"Error: {record.error}")

Next Run Preview

from philiprehberger_task_scheduler import Scheduler

scheduler = Scheduler()
scheduler.add("job", fn=my_function, cron="0 * * * *")

for name, next_time in scheduler.next_runs():
    print(f"{name}: next at {next_time}")

Task Dependencies

from philiprehberger_task_scheduler import Scheduler

scheduler = Scheduler()

@scheduler.interval(seconds=60, name="fetch-data")
def fetch_data():
    download_latest()

@scheduler.interval(seconds=60, name="process-data", depends_on="fetch-data")
def process_data():
    transform_and_store()

Graceful Shutdown

from philiprehberger_task_scheduler import Scheduler

scheduler = Scheduler()
scheduler.start(background=True)

# Wait for running tasks to finish before stopping
scheduler.stop(wait=True)

# Or set a timeout (seconds)
scheduler.stop(wait=True, timeout=10)

Missed Job Handling

from philiprehberger_task_scheduler import Scheduler, MissedJobPolicy

scheduler = Scheduler()

@scheduler.cron("0 * * * *", missed_policy=MissedJobPolicy.RUN_ONCE)
def hourly_sync():
    sync_data()

@scheduler.cron("*/5 * * * *", missed_policy=MissedJobPolicy.RUN_ALL)
def critical_check():
    check_systems()

Pause and Resume Jobs

from philiprehberger_task_scheduler import Scheduler

scheduler = Scheduler()
scheduler.add("my-job", fn=my_function, interval_seconds=60)

scheduler.pause("my-job")   # job won't run during ticks
scheduler.resume("my-job")  # job runs again

Job Management

from philiprehberger_task_scheduler import Scheduler

scheduler = Scheduler()
scheduler.add("job1", fn=fn1, interval_seconds=60)
scheduler.add("job2", fn=fn2, cron="0 * * * *")

print(scheduler.job_count)   # 2
print(scheduler.is_running)  # False

scheduler.clear()            # remove all jobs
print(scheduler.job_count)   # 0

Introspecting jobs

from philiprehberger_task_scheduler import Scheduler

scheduler = Scheduler()
scheduler.add("poller", fn=poll, interval_seconds=60)

print(scheduler.job_count)              # 1

job = scheduler.find_job("poller")
if job is not None:
    print(job.name, job.interval_seconds)  # poller 60

# find_job returns None on a miss — no exception to catch
assert scheduler.find_job("missing") is None

Cron Syntax

Standard 5-field cron expressions:

minute (0-59)
hour (0-23)
day of month (1-31)
month (1-12)
day of week (0-6, Mon-Sun)

Supports: *, ranges (1-5), lists (1,3,5), steps (*/5).

API

Development

pip install -e .
python -m pytest tests/ -v

Support

If you find this project useful:

⭐ Star the repo

🐛 Report issues

💡 Suggest features

❤️ Sponsor development

🌐 All Open Source Projects

💻 GitHub Profile

🔗 LinkedIn Profile

License

MIT