Scheduler
Cron-style routines with a swappable job store.
Cron-style routines, with the job store behind a protocol.
from agentino.scheduler import CronScheduler, FileJobStore
scheduler = CronScheduler(store=FileJobStore("data/jobs.json"))
await scheduler.start()
Stores
JobStore is a typing.Protocol. Four implementations ship:
| Store | Persists to | Use when |
|---|---|---|
InMemoryJobStore | nothing | Tests |
FileJobStore | one JSON file | A single process, and you want to read the jobs |
SqliteJobStore | a SQLite file | Many jobs, or concurrent access |
| Yours | anywhere | The jobs are already defined somewhere else |
That last case is more common than it sounds. If routines are declared in a YAML file that a person edits, the file is the truth and the store should read it rather than keep a second copy that can drift.
Retries
A recurring job that fails is retried with backoff, up to max_retries
(default 3), and then left alone until its next scheduled run. A job that fails
every time therefore does not spin — it fails three times an hour, not three
times a second.