DOCSEARCHbyAlgolia
Submit Docs

APScheduler Documentation

repository·master·Indexed 27 days ago

https://github.com/agronholm/apscheduler

A flexible in-process task scheduler and queue system for Python with Cron-like capabilities. It supports both synchronous and asynchronous environments, ranging from simple scripts to distributed multi-node production systems. The library includes various scheduler classes, job executors, data stores for persistence (SQLAlchemy, MongoDB, Memory), event brokers (Redis, MQTT, Asyncpg, Psycopg), and multiple trigger types including Cron, Interval, and Date triggers.

Tokens
12.3K
Snippets
24
Records
86
Agent score
92%

What's inside APScheduler

  1. Overview of APScheduler

    master
    Advanced Python Scheduler (APScheduler) is a task scheduler and task queue system for Python. It supports both synchronous (thread-based) and asynchronous (asyncio or Trio) applications. It is designed to scale from single-process use cases to large, multi-node deployments using shared data stores for high availability and horizontal scaling.

  2. Core concepts of APScheduler

    master

    Understanding the fundamental components:

    • Callable: Any object that returns True from callable() (e.g., functions, methods, lambdas, or classes with a __call__ method).
    • Task: Encapsulates a callable and configuration parameters. It provides the target code, a task ID for concurrency limiting, and a template for parameters like job executor and misfire grace time.
    • Trigger: Contains the logic to calculate when a task should run.
    • Schedule: Combines a Task with a Trigger and configuration parameters.
    • Job: A request for a task to be run. Created automatically from a schedule or directly by the user.
    • Data Store: Stores schedules, jobs, and tracks tasks.
    • Job Executor: Runs the job by calling the associated callable (can be in a thread, subprocess, or external service).
    • Event Broker: Facilitates cooperation between schedulers by notifying them of new/updated schedules and jobs.
    • Scheduler: The main interface. It houses the data store, event broker, and job executors. It processes schedules to spawn jobs and runs available jobs.

  3. Run tasks without scheduling

    master

    You can use the scheduler as a job queue to run tasks directly without a schedule.

    • To queue a job and wait for its completion and result, use Scheduler.run_job.
    • To launch a job without waiting for its result, use Scheduler.add_job.
    • To retrieve results later, pass a result_expiration_time to Scheduler.add_job to ensure the result is saved, then use Scheduler.get_job_result with the job ID returned by add_job.

  5. Schedule tasks using triggers

    master

    To schedule a task, you need a task (or a callable) and a trigger.

    Built-in Triggers:

    • DateTrigger: Run once at a specific time.
    • IntervalTrigger: Run at fixed intervals.
    • CronTrigger: Run periodically at specific times (e.g., daily at 10:00).
    • CalendarIntervalTrigger: Run on calendar-based intervals (e.g., every 2 months).
    from apscheduler.triggers.combining import OrTrigger
from apscheduler.triggers.cron import CronTrigger

# Example: Combining triggers with OrTrigger
trigger = OrTrigger(
    CronTrigger(day_of_week="mon-fri", hour=10),
    CronTrigger(day_of_week="sat-sun", hour=11),
)

  6. Understand settings inheritance for tasks and jobs

    master

    Settings (like job_executor or misfire_grace_time) follow a specific priority order. If a parameter is unset at one level, it is looked up in the parent level.

    Priority Order for Task Configuration:

    1. Parameters passed directly to AsyncScheduler.configure_task.
    2. Parameters bound via the @task decorator.
    3. The scheduler's TaskDefaults.

    Inheritance Flow:

    • Schedules inherit from their tasks.
    • Jobs created from schedules inherit from their parent schedules.
    • Jobs created directly inherit from their parent tasks.

    Metadata Inheritance: Metadata keys are merged, with more explicit configurations overwriting generic ones.

  8. Run an Asynchronous Scheduler

    master

    If your application runs on asyncio or Trio, use AsyncScheduler. You must use the scheduler as an asynchronous context manager (async with).

    Run in foreground (blocking)

    Use await scheduler.run_until_stopped() to block the event loop.

    Run in background

    Use await scheduler.start_in_background() to run the scheduler as a background task within the event loop.

    import asyncio
from apscheduler import AsyncScheduler

async def main():
    # Background task mode
    async with AsyncScheduler() as scheduler:
        # Add schedules, configure tasks here
        await scheduler.start_in_background()
    
    # OR Foreground blocking mode
    # async with AsyncScheduler() as scheduler:
    #     await scheduler.run_until_stopped()

asyncio.run(main())

  9. Assign a custom identity to the scheduler

    master

    When running in production with persistent stores, assign a unique, stable identity to each scheduler instance. This helps identify which jobs are running where and allows crashed jobs to be cleaned up more efficiently by other schedulers.

    • Best practice: Use a value that is unique among instances but remains the same upon restart (e.g., a Kubernetes Pod name).
    • Single instance: Use a static scheduler ID.
    • Default behavior: If no ID is provided, APScheduler generates one using the hostname, process ID, and instance ID.

  10. Integrate APScheduler with WSGI frameworks

    master

    To integrate APScheduler with WSGI-based web frameworks, use the synchronous Scheduler class. You should start the scheduler as a side effect of importing the module containing your application instance.

    When using uWSGI, you must include the --enable-threads (or -T) flag in your command, as uWSGI disables threads by default, which will prevent the scheduler from running.

    Note that calling Scheduler.start_in_background() installs an atexit hook to ensure the scheduler shuts down gracefully when the worker process exits.

    from apscheduler import Scheduler


def app(environ, start_response):
    """Trivial example of a WSGI application."""
    response_body = b"Hello, World!"
    response_headers = [
        ("Content-Type", "text/plain"),
        ("Content-Length", str(len(response_body))),
    ]
    start_response(200, response_headers)
    return [response_body]

scheduler = Scheduler()
scheduler.start_in_background()

    To run with uWSGI:

    uwsgi --enable-threads --http :8080 --wsgi-file example.py

  11. Migrate from v3.x to v4.0: Scheduler API Changes

    master

    The scheduler API has changed significantly in v4.0:

    • Adding Jobs: Use Scheduler.add_schedule() for scheduled tasks. The add_job() method is now reserved for one-off runs (previously requiring a DateTrigger with the current time).
    • Unified Scheduler: BlockingScheduler and BackgroundScheduler are merged into a single Scheduler class.
      • Use Scheduler.run_until_stopped() to replace BlockingScheduler behavior.
      • Use Scheduler.start_in_background() to replace BackgroundScheduler behavior.
    • Async Support: AsyncScheduler replaces the old asyncio scheduler and is based on AnyIO (supporting both asyncio and Trio). Note: AsyncScheduler must be used as an async context manager.
    • Configuration: The configure() method has been removed. All configuration is now passed as keyword arguments directly to the scheduler class.
    • Data Stores: Schedulers no longer support multiple data stores; run multiple schedulers if this capability is needed.

  12. Install APScheduler

    master

    Install the core library using pip:

    $ pip install apscheduler

    To ensure compatibility with external services, you can install specific extras. This is preferred over installing the libraries separately. You can install multiple extras by providing them as a comma-separated list inside brackets.

    pip install apscheduler