Core Concepts
Defining Jobs
Registering jobs, naming and tagging them, and controlling retries and timeouts.
The decorator
@crons.cron() registers the decorated function and returns it unchanged, so you can still call it directly from your own code or tests.
@crons.cron("0 */2 * * *", name="data_sync", tags=["sync", "data"])
async def sync_data():
# Runs every 2 hours.
await pull_upstream_changes()
return "synced"Parameters
None falls back to CRON_DEFAULT_MAX_RETRIES.None retries on any exception.JobTimeoutError.Names are identity
State, locks and history are all keyed by job name. Renaming a job orphans its previous history and starts it fresh — set name= explicitly if you expect to rename the underlying function.
Sync and async jobs
Both are supported. Async jobs are awaited on the event loop. Sync jobs are dispatched to a thread, so a slow blocking call will not stall your API.
@crons.cron("*/10 * * * *")
async def async_job():
async with httpx.AsyncClient() as client:
await client.get("https://api.example.com/ping")
@crons.cron("*/10 * * * *")
def blocking_job():
# Runs in a thread, so blocking here is safe.
subprocess.run(["./backup.sh"], check=True)Tags
Tags are returned by the API and passed to hooks in the execution context, which makes them a convenient way to apply behaviour to a whole class of jobs.
@crons.cron("0 2 * * *", tags=["backup", "critical"])
async def backup_database():
...
def alert_on_critical(job_name: str, context: dict):
if "critical" in context["tags"]:
page_on_call(job_name, context)
crons.add_on_error_hook(alert_on_critical)Retries
Set max_retries to retry a failing job. Delays grow exponentially and include jitter, so a fleet of replicas does not retry in lockstep.
@crons.cron(
"*/10 * * * *",
name="sync_api",
max_retries=3,
retry_delay=5.0,
retry_on=(ConnectionError, TimeoutError),
)
async def sync_external_api():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.example.com/orders")
response.raise_for_status()
return response.json()The delay before attempt n is:
delay = min(retry_delay * (backoff_multiplier ** n) + jitter, max_delay)With the defaults — a multiplier of 2.0 and a 300 second ceiling — that gives roughly 1s, 2s, 4s, 8s and so on. All four values are configurable globally through environment variables.
Retries hold the job slot
A job retrying with long backoff is still running. If the next scheduled tick arrives first, that tick is skipped rather than overlapped — set timeout to bound the total time a job can occupy.
Timeouts
timeout caps how long a single attempt may run. Exceeding it raises JobTimeoutError, which triggers the error hooks and counts as a failed attempt for retry purposes.
from fastapi_crons import JobTimeoutError
@crons.cron("*/30 * * * *", timeout=120)
async def slow_report():
await build_monthly_report()Sync jobs cannot be interrupted
A timeout stops waiting on the job and marks the run failed, but a blocking call already running in a thread keeps going until it returns. Prefer async jobs, or pass a timeout down to whatever you are calling, when this matters.
Retrying ordinary functions
retry_on_failure applies the same retry machinery to any callable, whether or not it is a cron job.
from fastapi_crons import retry_on_failure
@retry_on_failure(max_retries=3, retry_delay=1.0)
async def fetch_data():
response = await http_client.get("https://api.example.com")
return response.json()Running a job by hand
Jobs can be triggered outside their schedule — over HTTP:
curl -X POST http://127.0.0.1:8000/api/backup_database/runor from the CLI:
python -m fastapi_crons.cli run-job backup_databaseManual runs set manual_trigger to true in the hook context, so you can tell them apart from scheduled runs.