Advanced Python Scheduler (APScheduler)

3.11.2 · active · verified Sun Mar 29

APScheduler is a flexible, in-process task scheduler library for Python, offering cron-like capabilities. It allows you to schedule Python code to be executed later, either once or periodically, within your application. The current stable version is 3.11.2. It supports various scheduler types, job stores (e.g., in-memory, SQLAlchemy, MongoDB), and triggers (date, interval, cron, calendarinterval). APScheduler is primarily meant to be run inside existing applications, not as a standalone daemon. It is actively maintained with a stable 3.x series and an ongoing pre-release 4.x series with significant architectural changes.

Warnings

breaking APScheduler 4.0 (currently in pre-release) introduces significant breaking changes. Key changes include a split of the 'job' concept into 'Task', 'Schedule', and 'Job', the merging of 'workers' into 'schedulers', removal of synchronous interfaces for event brokers and data stores, and a completely overhauled job store system that is incompatible with 3.x data. The `add_job()` method is renamed to `add_schedule()`.
Fix: Refer to the APScheduler 4.x migration guide before upgrading. A direct migration path for persistent job store data from 3.x to 4.x is not automatically available at the time of writing and may require manual recreation or a specific migration tool when 4.x is stable.
gotcha When using `BackgroundScheduler`, if your main script finishes, the scheduler will also stop. You need to keep the main thread alive (e.g., with `time.sleep()`, a web server, or another blocking call) for the background jobs to execute.
Fix: For standalone scripts, use `while True: time.sleep(interval)` with a `try...except KeyboardInterrupt` block, or use `BlockingScheduler` if the scheduler is the only thing running in your process. For web applications, ensure the scheduler's lifecycle is tied to the application's.
gotcha Functions scheduled with persistent job stores (e.g., SQLAlchemyJobStore, MongoDBJobStore) must be importable by a textual reference (e.g., 'mymodule:my_function'). Lambda functions, bound methods, or nested functions often cannot be properly serialized and deserialized across application restarts or different processes, leading to `ValueError`.
Fix: Ensure scheduled functions are top-level functions in a module, or static/class methods, and provide their fully qualified path (e.g., `scheduler.add_job('my_module.my_function', ...)`).
gotcha Sharing a persistent job store among multiple APScheduler *instances* (in different processes or nodes) directly can lead to incorrect behavior like duplicate job execution or missed jobs, as APScheduler 3.x does not have built-in interprocess synchronization for job stores.
Fix: For multi-process or multi-node deployments, APScheduler 3.x typically requires a single, dedicated scheduler process that manages jobs, with other processes communicating with it. Alternatively, consider using a distributed task queue (like Celery) if strong guarantees and distributed coordination are critical. APScheduler 4.x aims to address this with enhanced data stores and event brokers.
deprecated Support for `pytz` time zones has been deprecated in version 3.11.0 in favor of `zoneinfo` (or `backports.zoneinfo` for Python < 3.9).
Fix: Migrate your timezone definitions to use `zoneinfo` (built-in in Python 3.9+) or `backports.zoneinfo` for older Python versions. For example, `from zoneinfo import ZoneInfo` instead of `from pytz import timezone`.

Install

pip install apscheduler Core library
pip install apscheduler[sqlalchemy] With SQLAlchemy job store
pip install apscheduler[mongodb] With MongoDB job store
pip install apscheduler[redis] With Redis job store/event broker
pip install apscheduler[asyncpg] With AsyncPG event broker

Imports

BackgroundScheduler

from apscheduler.schedulers.background import BackgroundScheduler

BlockingScheduler

from apscheduler.schedulers.blocking import BlockingScheduler

AsyncIOScheduler

from apscheduler.schedulers.asyncio import AsyncIOScheduler

IntervalTrigger

from apscheduler.triggers.interval import IntervalTrigger

CronTrigger

from apscheduler.triggers.cron import CronTrigger

DateTrigger

from apscheduler.triggers.date import DateTrigger

Quickstart

This quickstart demonstrates a `BackgroundScheduler` which runs jobs in a separate thread without blocking the main application. It schedules a simple function to run every 5 seconds using an `IntervalTrigger`. The `while True: time.sleep(2)` loop keeps the main thread alive, allowing the background scheduler to operate. A `KeyboardInterrupt` handler ensures a graceful shutdown of the scheduler.

import time
from datetime import datetime
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger

def my_job():
    print(f"Hello from APScheduler! The time is: {datetime.now()}")

if __name__ == '__main__':
    scheduler = BackgroundScheduler()
    scheduler.add_job(my_job, IntervalTrigger(seconds=5), id='my_scheduled_job')
    
    print('Starting scheduler. Press Ctrl+C to exit.')
    scheduler.start()

    try:
        # This is here to simulate application activity (for BackgroundScheduler)
        while True:
            time.sleep(2)
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()
        print("Scheduler shut down successfully.")

view raw JSON →