apscheduler
APScheduler runs Python functions on a schedule from inside your own process. You create a scheduler object, add jobs with a trigger (cron expression, fixed interval, calendar interval, or a single future datetime), and it keeps a timer thread or event loop task awake to fire them. There is no broker, no worker fleet, and no separate daemon: the scheduler lives in the same interpreter as your web app or script. Job state can optionally be persisted to SQLAlchemy, MongoDB, Redis, or ZooKeeper so schedules survive a restart.
The right answer when one Python process needs richer scheduling than cron and you do not want to run a queue. The moment you have more than one replica of that process, or need a job to be retried, it stops being enough.
Use it if
- You want cron-like scheduling inside a long-running Python service without adding a broker and worker processes
- You need triggers that plain cron cannot express, such as every 90 seconds, the third of every month at a fixed local time, or a combination of both
- You are writing an asyncio or Trio application and want a scheduler that runs on the event loop rather than in a thread pool
- You need schedules that survive process restarts and can be stored in a database you already run
- You run multiple web workers: gunicorn with four workers starts four schedulers and fires every job four times unless you add a lock or move the scheduler into its own single process
- You need retries, result storage, or fan-out across machines: APScheduler has none of that, and Celery, Dramatiq, or arq exist for exactly that gap
- You need work isolated from request handling: jobs execute in a thread or process pool inside your app, so a slow job competes with your HTTP traffic for the same interpreter
- You read the GitHub README and got excited about the 4.x feature list: 4.0 has been in alpha since 2022, the newest pre-release is 4.0.0a6 from April 2025, and the README itself says do not use it in production; the stable package is 3.11.x with a different API
- A missed run is unacceptable: if the process is down when a job was due, the default misfire_grace_time drops it silently rather than running it late
Setup reality
pip install apscheduler pulls almost nothing (tzlocal is the only hard dependency), and the first BackgroundScheduler takes four lines. The friction shows up after that. You pick a scheduler class per runtime (BackgroundScheduler, BlockingScheduler, AsyncIOScheduler, plus gevent, Tornado, Twisted and Qt variants) and persistence needs an extra: apscheduler[sqlalchemy], [redis], or [mongodb]. Persistent jobs are stored as a pickled reference to module:function, so renaming or moving a scheduled function orphans the rows already in the store. Every add_job in module-level or startup code needs a stable id plus replace_existing=True or each restart quietly stacks another copy. 3.11 also deprecated pytz time zones in favor of ZoneInfo, so older tutorials produce warnings.
Patterns
Start a scheduler inside a running appbackground-scheduler
from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler(timezone="UTC")
scheduler.start()
# keep a reference; if it is garbage collected the timer thread stopsBackgroundScheduler runs in a daemon thread, so it dies with the process and nothing is retried. Call scheduler.shutdown() on shutdown or in-flight jobs get cut off.
Run a function every N minutesinterval-job
def sync_inventory():
...
scheduler.add_job(
sync_inventory,
"interval",
minutes=5,
id="sync-inventory",
replace_existing=True,
)Without a fixed id plus replace_existing=True, a persistent job store gains a duplicate job on every restart. The first run happens one interval from now unless you pass next_run_time.
Schedule with a cron-style triggercron-job
from apscheduler.triggers.cron import CronTrigger
scheduler.add_job(
nightly_report,
CronTrigger(hour=3, minute=30, timezone="Asia/Kolkata"),
id="nightly-report",
replace_existing=True,
)
# or from a crontab string
scheduler.add_job(nightly_report, CronTrigger.from_crontab("30 3 * * *"))Any cron field you leave out defaults to * for the fields below the smallest one you set, which is why add_job(f, "cron", hour=3) fires once a day and not once a minute during hour 3.
Run once at a specific timeone-off-job
from datetime import datetime, timedelta, timezone
scheduler.add_job(
send_reminder,
"date",
run_date=datetime.now(timezone.utc) + timedelta(hours=2),
args=[user_id],
)A date job is removed from the store once it has run. Pass timezone-aware datetimes; naive ones are interpreted in the scheduler time zone, which is rarely what you meant on a UTC server.
Schedule coroutines on the event loopasyncio-scheduler
import asyncio
from apscheduler.schedulers.asyncio import AsyncIOScheduler
async def poll_api():
...
async def main():
scheduler = AsyncIOScheduler()
scheduler.add_job(poll_api, "interval", seconds=30)
scheduler.start()
await asyncio.Event().wait()
asyncio.run(main())AsyncIOScheduler runs coroutine jobs directly on the loop, so a blocking call inside one stalls the whole application. Push CPU-bound work to a ProcessPoolExecutor job instead.
Persist schedules to a databasepersistent-jobstore
# pip install "apscheduler[sqlalchemy]"
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
scheduler = BackgroundScheduler(
jobstores={"default": SQLAlchemyJobStore(url="postgresql+psycopg://user:pw@db/app")},
timezone="UTC",
)Stored jobs keep a pickled reference to module:function. Move or rename the function and the job fails to load on the next start, so cleanup is a migration step you have to remember.
Control overlap, misfires, and the executor pooljob-defaults
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
scheduler = BackgroundScheduler(
executors={
"default": ThreadPoolExecutor(10),
"cpu": ProcessPoolExecutor(4),
},
job_defaults={
"coalesce": True,
"max_instances": 1,
"misfire_grace_time": 300,
},
)
scheduler.add_job(crunch_numbers, "cron", hour="*", executor="cpu")max_instances=1 is what stops a slow five-minute job from stacking on top of itself. coalesce=True collapses a backlog of missed runs into one, and misfire_grace_time decides how late is still worth running.
Pause, reschedule, and remove jobs at runtimemanage-jobs
job = scheduler.get_job("sync-inventory")
job.pause()
job.resume()
job.modify(args=[new_arg], name="sync inventory v2")
job.reschedule(trigger="interval", minutes=15)
scheduler.remove_job("sync-inventory")
scheduler.print_jobs()modify() changes job attributes; changing the schedule needs reschedule(). Calling remove_job with an id that is not there raises JobLookupError rather than passing quietly.
Get told when a job fails or is missederror-listener
import logging
from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_MISSED
log = logging.getLogger(__name__)
def on_job_event(event):
if event.exception:
log.error("job %s raised", event.job_id, exc_info=event.exception)
else:
log.warning("job %s missed its run time", event.job_id)
scheduler.add_listener(on_job_event, EVENT_JOB_ERROR | EVENT_JOB_MISSED)Exceptions inside jobs are swallowed by the executor and only surface through the apscheduler logger or a listener, so an unmonitored scheduler fails completely silently.
Run on calendar intervals at a fixed wall clock timecalendar-interval
from apscheduler.triggers.calendarinterval import CalendarIntervalTrigger
scheduler.add_job(
monthly_invoice,
CalendarIntervalTrigger(months=1, hour=9, timezone="Europe/Berlin"),
id="monthly-invoice",
replace_existing=True,
)Backported from the 4.x branch in 3.11.0. The docstring warns that a start date on the 29th to 31st skips months where that day does not exist, and that a run time inside the DST switch window can be skipped or repeated.
Combine triggers with AND or ORcombining-triggers
from apscheduler.triggers.combining import AndTrigger, OrTrigger
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
# every 2 hours, but only on weekdays
scheduler.add_job(
poll_feed,
AndTrigger([IntervalTrigger(hours=2), CronTrigger(day_of_week="mon-fri")]),
)
# 08:00 and 18:00
scheduler.add_job(digest, OrTrigger([CronTrigger(hour=8), CronTrigger(hour=18)]))AndTrigger only fires when all sub-triggers agree on the same fire time, so combining two intervals that never line up produces a job that never runs.
Block on the scheduler and shut down cleanlygraceful-shutdown
from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler(timezone="UTC")
scheduler.add_job(heartbeat, "interval", seconds=60)
try:
scheduler.start() # blocks here
except (KeyboardInterrupt, SystemExit):
scheduler.shutdown(wait=True)BlockingScheduler is the right class for a dedicated scheduler container, which is also how you avoid every gunicorn worker running its own copy of the schedule.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| celery | PyPI | You need retries, routing, and workers on other machines, and can operate a broker to get them. |
| schedule | PyPI | A single script needs a human-readable timer loop and you do not care about persistence or time zones. |
| arq | PyPI | Your app is asyncio and you want cron jobs plus a real Redis-backed queue with retries in one library. |