🦀 Rust Core 🐍 Python API v0.1.5 📦 PyPI

Schedule jobs in Python,
at Rust speed

Interval and cron jobs for FastAPI, Django, Celery, and any Python backend.
Run them on a background thread with per-job retries and live inspection — no database, no broker.

$ pip install rust-py-scheduler
Get Started View on GitHub
python
>>> from rust_py_scheduler import Scheduler
>>> scheduler = Scheduler()
>>> scheduler.every("30s", refresh_cache)
'a1b2c3d4-...' # job id
>>> scheduler.cron("0 9 * * 1-5", send_report)
>>> scheduler.start_background()
>>> █
103Tests passing
cron + intervalSchedules
3Integrations
0Runtime deps

Everything a scheduler needs

Rust handles timing, threads, and retries. Python stays simple.

⏱️

Interval Scheduling

Run a job "10s", "5m", or "1h" apart. One readable string — no timestamps to manage.

📅

Cron Expressions

Standard 5-field Unix cron. scheduler.cron("0 9 * * 1-5", fn) runs weekdays at 9am, in your local timezone.

🎀

Call or Decorate

Register with a direct call (scheduler.every("5s", fn)) or a decorator (@scheduler.every("5s")) — your function stays callable.

🧵

Background Thread

start_background() runs the loop on a dedicated OS thread and returns immediately. The GIL is released while idle.

🔁

Per-Job Retries

max_retries=N retries a failing job immediately, up to N extra times, before the tick counts as an error.

🛡️

Jobs Never Crash the Loop

An exception in one job is caught, printed, and tracked in error_count / last_error. Every other job keeps running.

⚡

FastAPI & Django

scheduler_lifespan() for FastAPI, start_in_background() for Django — start and stop with your app's lifecycle.

🥬

Celery-friendly

A task's .delay is just a callable, so scheduler.every("5m", task.delay) works with no extra glue.

🦀

Rust Core via PyO3

Scheduling, timing, retries, and thread management compile to a native .so extension. No Rust needed to use it.

Works with your stack

Call the core API, or wire it into your framework's lifecycle.

Python 
import time
from rust_py_scheduler import Scheduler

scheduler = Scheduler()

# Direct call: registers immediately, returns the job id.
job_id = scheduler.every("2s", lambda: print("tick"))

# Decorator form: same registration, function stays callable.
@scheduler.every("3s", max_retries=2)
def report():
    print("tick (with retry budget)")

scheduler.start_background()  # returns immediately
time.sleep(7)

for job in scheduler.list_jobs():
    print(job)  # run_count, error_count, next_run_at, ...

scheduler.remove_job(job_id)
scheduler.shutdown()
Python 
from rust_py_scheduler import Scheduler

scheduler = Scheduler()

# 5-field Unix cron: minute hour day-of-month month day-of-week
scheduler.cron("0 * * * *", hourly_task)      # top of every hour
scheduler.cron("*/15 * * * *", poll)          # every 15 minutes

@scheduler.cron("0 9 * * 1-5")              # weekdays at 9am
def morning_report():
    ...

@scheduler.cron("30 2 * * *", max_retries=2)  # daily at 02:30
def nightly_cleanup():
    ...

# Evaluated in local time. Invalid expressions raise ValueError now,
# not later — fields support *, a-b, */step, and 1,2,3 lists.
Python 
from fastapi import FastAPI
from rust_py_scheduler import Scheduler
from rust_py_scheduler.fastapi import scheduler_lifespan

scheduler = Scheduler()

@scheduler.every("30s")
def heartbeat():
    ...

# Starts on app startup, shuts down (and joins) on app shutdown.
app = FastAPI(lifespan=scheduler_lifespan(scheduler))
Python 
# apps.py
from django.apps import AppConfig
from rust_py_scheduler import Scheduler
from rust_py_scheduler.django import start_in_background

scheduler = Scheduler()

@scheduler.every("5m")
def refresh_cache():
    ...

class MyAppConfig(AppConfig):
    name = "myapp"

    def ready(self):
        # Idempotent per process; best-effort atexit shutdown.
        start_in_background(scheduler)
Python 
from rust_py_scheduler import Scheduler

scheduler = Scheduler()

# A Celery task's .delay is just a callable — no extra glue needed.
scheduler.every("5m", lambda: send_report.delay("daily-metrics"))
scheduler.cron("0 8 * * 1-5", lambda: send_report.delay("digest"))

# Need countdown / eta / queue routing? Use apply_async:
scheduler.every(
    "1h",
    lambda: send_report.apply_async(args=["hourly"], countdown=10),
)

# The scheduler only enqueues; the Celery worker does the work.

Cron, the standard way

Five fields, Unix semantics — the expressions you already know.

minute
0–59
*/15, 0,30
·
hour
0–23
9-17
·
day of month
1–31
1, 15
·
month
1–12
*
·
day of week
0–7 (Sun=0/7)
1-5

Field syntax

  • * every value
  • 5 single
  • 9-17 range
  • */15 step
  • 0,30 list

Good to know

  • local time timezone
  • 1 minute resolution
  • ValueError eager validation
scheduler.cron("0 9 * * 1-5", fn) 
# weekdays at 9am — registered, returns a job id
"7f3c9a12-4b8e-..."
scheduler.list_jobs()[0] 
{
  "schedule": "cron 0 9 * * 1-5",
  "run_count": 3,
  "next_run_at": "1718960400",
  "last_error": None
}

How it works

Rust schedules and runs. Python registers. You inspect.

Your App
FastAPI Django Celery Script
→
Python API
every() / cron() start_background() list_jobs()
→
🦀 Rust Core
scheduler.rs registry.rs executor.rs cron.rs interval.rs
→
Output
jobs run on time retries + counts list[dict] state

The native .so extension is compiled once at publish time via maturin + PyO3, against the stable ABI (abi3-py310) — one wheel covers Python 3.10–3.13+.
Your users just pip install — no Rust toolchain, no database, no message broker required.

Ready in seconds

Install from PyPI. No Rust, no compilers, no configuration.

Basic

pip install rust-py-scheduler

With FastAPI

pip install "rust-py-scheduler[fastapi]"

With Django

pip install "rust-py-scheduler[django]"

With Celery

pip install "rust-py-scheduler[celery]"
Requires Python 3.10+ · Linux · macOS · Windows · MIT License
Copied!