schedium

A lightweight, composable, in-process, pure-python job scheduler.

Why schedium?

Most Python schedulers either require a background thread / daemon or force you into a rigid cron syntax. schedium takes a different approach:

• No threads, no processes — jobs run inline when you call run_pending(). They can be run in threads or asynchronously using helpers.
• Composable triggers — build complex schedules by combining simple primitives with & (AND) and | (OR). See composing triggers doc
• Automatic deduplication — calling run_pending() multiple times within the same time bucket is safe; jobs run at most once per bucket. See deduplication doc
• Zero dependencies — pure Python, nothing outside the standard library.
• Fully typed — first-class type annotations and mypy-checked.
• Supports all currently maintained Python versions: 3.10, 3.11, 3.12, 3.13, and 3.14.

link to the documentation

Installation

pip install schedium

Quick start

import time
from schedium import Every, Job, Scheduler, Weekly

sched = Scheduler()

def hello():
    print("hello!")

# Every 5 minutes
sched.append(Job(hello, Every(unit="minute", interval=5), name="5-min"))

# Every Monday at 09:30
sched.append(Job(hello, Weekly("monday", at="09:30"), name="weekly"))

while True:
    sched.run_pending()
    time.sleep(1)

Threading (optional)

schedium runs jobs inline by default. If you want multi-threading, use the helpers in schedium.threading:

• ThreadedJobsScheduler: runs each due job on a worker thread (thread pool).
• QueuedJobsScheduler: keeps the scheduler in your thread and enqueues due jobs for worker threads.
• SchedulerThread: runs the scheduler loop itself in a dedicated thread.

from schedium import Every, Job, Scheduler
from schedium.threading import SchedulerThread, ThreadedJobsScheduler

sched = Scheduler()
sched.append(Job(lambda: print("tick"), Every(unit="second", interval=1)))

threaded = ThreadedJobsScheduler(sched, max_workers=8)
runner = SchedulerThread(threaded, interval=1.0)
runner.start()

# ... later
runner.stop()
runner.join()
threaded.shutdown()

Async (optional)

For asyncio applications, use schedium.asyncio.AsyncScheduler. It is used the same way as Scheduler, but runs due jobs on the event loop — async def jobs are awaited directly, plain sync jobs run in an executor so they never block the loop.

import asyncio
from schedium import Every, Job
from schedium.asyncio import AsyncScheduler

async_sched = AsyncScheduler()
async_sched.append(Job(lambda: print("tick"), Every(unit="second", interval=1)))

async def main():
    while True:
        await async_sched.run_pending()
        await asyncio.sleep(1)

asyncio.run(main())

Composing triggers

Triggers are the building blocks of schedules. Combine them freely:

from schedium import Every, On, Between

# Every minute, but only on weekdays between 9 AM and 5 PM
trigger = (
    Every(unit="minute", interval=1)
    & On(unit="weekdays")
    & Between(unit="hour_of_day", start=9, end=17)
)

from schedium import Every, On

# Every hour, at minute 12 OR minute 55
trigger = (
    Every(unit="hour", interval=1)
    & (On(unit="minute_of_hour", value=12) | On(unit="minute_of_hour", value=55))
)

Available triggers

Deduplication

schedium automatically deduplicates job runs. Calling run_pending() repeatedly within the same time bucket will only execute the job once:

from datetime import datetime
from schedium import JobDidNotRun, Every, Job, Scheduler

sched = Scheduler()
sched.append(Job(lambda: print("tick"), Every(unit="minute", interval=1)))

# First call at 10:05 → runs the job
sched.run_pending(now=datetime(2026, 2, 4, 10, 5, 0))

# Second call at 10:05 → already ran for this bucket
result = sched.run_pending(now=datetime(2026, 2, 4, 10, 5, 0))
assert result[0] is JobDidNotRun

# Next minute → runs again
sched.run_pending(now=datetime(2026, 2, 4, 10, 6, 0))

Inspecting next run times

from datetime import datetime
from schedium import Every, Job, Scheduler

sched = Scheduler()
sched.append(Job(lambda: None, Every(unit="minute", interval=5)))

next_run = sched.time_of_next_run(after=datetime(2026, 2, 4, 10, 3, 0))
print(next_run)  # datetime(2026, 2, 4, 10, 5, 0)

Timezone handling

For predictable behavior, use UTC-aware datetimes:

import time
from datetime import datetime, timezone
from schedium import Every, Job, Scheduler

sched = Scheduler()
sched.append(Job(lambda: print("tick"), Every(unit="minute", interval=1)))

while True:
    sched.run_pending(now=datetime.now(timezone.utc))
    time.sleep(1)

Local timezones work too (via zoneinfo), but be aware of DST transitions — see the docs for details.

Documentation

Full documentation is built with Sphinx and hosted alongside the project:

• Guides: Scheduler usage, job creation, trigger composition
• Concepts: Granularity, trigger tokens & deduplication, window time
• API reference: Every class and function documented with numpydoc

Build locally:

pip install -e . --group docs
sphinx-build -b html docs docs/_build/html

Development

Setup

git clone https://github.com/MarcBresson/schedium.git
cd schedium
python -m venv .venv && source .venv/bin/activate
pip install -e . --group dev --group test --group docs
pre-commit install

Run tests

pytest

Linting & formatting

The project uses Ruff for linting and formatting, mypy for type checking, and numpydoc for docstring validation — all enforced via pre-commit:

pre-commit run --all-files

Contributing

Contributions are welcome! Please:

1. Fork the repository
2. Create a feature branch (git switch -c feature/my-feature)
3. Ensure all tests pass (pytest) and pre-commit hooks are clean
4. Open a pull request

License

schedium is licensed under the Apache License 2.0.