mrkeyoor.com_
Thu 06 Aug 15:43 UTC
PyPIWeb Backendupdated 06 Aug 2026

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.

Verdict

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.

API stability4/5The 3.x API has not moved in years and 3.11.x releases are DST and bugfix only, but the long-running 4.0 alpha means the project README and the master docs describe an API that pip will not give you.
Docs4/5The 3.x ReadTheDocs branch has a proper user guide, a full trigger and job store reference, and runnable examples per scheduler type; the trap is that search engines often land you on the 4.x master docs instead.
Maintenance4/5Pushed within a week of this review by a single very active maintainer, with 40 open issues and 3.11.3 shipped in June 2026 fixing real DST bugs; the bus factor and the stalled 4.0 line are the risks.
Ecosystem4/5Job stores for SQLAlchemy, MongoDB, Redis, RethinkDB, etcd, and ZooKeeper, executors for threads, processes, asyncio, gevent, Tornado, and Twisted, plus long-standing third-party glue such as django-apscheduler and flask-apscheduler.

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
Skip it if

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 stops

BackgroundScheduler 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

PackageRegistryPick it when
celeryPyPIYou need retries, routing, and workers on other machines, and can operate a broker to get them.
schedulePyPIA single script needs a human-readable timer loop and you do not care about persistence or time zones.
arqPyPIYour app is asyncio and you want cron jobs plus a real Redis-backed queue with retries in one library.