Python APScheduler

A lightweight, in-process task scheduler for Python with cron-like capabilities, persistent job storage, and multiple execution backends.

Overview

APScheduler (Advanced Python Scheduler) enables scheduling Python functions to run at specific times or intervals. It supports multiple trigger types (date, interval, cron), various job stores (memory, database, Redis), and different executors (thread pool, process pool, asyncio). It's ideal for background tasks, periodic jobs, and cron-like scheduling within Python applications.

Version note: This sheet covers APScheduler 3.x (current stable PyPI release, 3.11.x). APScheduler 4.x is a pre-release major rewrite with a completely different API (AsyncScheduler, data stores instead of job stores, AnyIO-based). Do not use 4.x in production; do not mix these APIs.

flowchart TB
    subgraph APScheduler Architecture
        A[Scheduler] --> B[Job Store]
        A --> C[Executor]
        A --> D[Triggers]

        B --> E[(Memory)]
        B --> F[(SQLAlchemy)]
        B --> G[(Redis)]
        B --> H[(MongoDB)]

        C --> I[ThreadPool]
        C --> J[ProcessPool]
        C --> K[AsyncIO]

        D --> L[Date]
        D --> M[Interval]
        D --> N[Cron]
    end

    O[Job Function] --> A
    A --> P[Execute at Scheduled Time]

Job Scheduling Basics

APScheduler provides three scheduler classes for different use cases and execution environments.

Key Concepts

Scheduler: The main component that manages jobs, triggers, and executors
Job: A scheduled task containing a function reference and execution parameters
Trigger: Determines when a job should run (date, interval, or cron)
Job Store: Persists job data between scheduler restarts
Executor: Runs the actual job functions

Scheduler Types

# BlockingScheduler - Blocks the main thread (standalone scripts)
from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler()

# BackgroundScheduler - Runs in background thread (applications)
from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler()

# AsyncIOScheduler - For asyncio applications
from apscheduler.schedulers.asyncio import AsyncIOScheduler
scheduler = AsyncIOScheduler()

# GeventScheduler - For gevent applications
from apscheduler.schedulers.gevent import GeventScheduler

# TornadoScheduler - For Tornado applications
from apscheduler.schedulers.tornado import TornadoScheduler

Basic Usage

from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime

def my_job():
    print(f"Job executed at {datetime.now()}")

# Create and configure scheduler
scheduler = BackgroundScheduler()

# Add a job
scheduler.add_job(my_job, 'interval', seconds=30)

# Start the scheduler
scheduler.start()

# Keep the main thread alive (for background scheduler)
try:
    while True:
        pass
except KeyboardInterrupt:
    scheduler.shutdown()

Scheduler Lifecycle

stateDiagram-v2
    [*] --> Created : BackgroundScheduler()
    Created --> Running : start()
    Running --> Paused : pause()
    Paused --> Running : resume()
    Running --> Stopped : shutdown()
    Paused --> Stopped : shutdown()
    Stopped --> [*]

Trigger Types

Triggers determine when and how often jobs execute. APScheduler provides three built-in trigger types.

Date Trigger

Runs once at a specific date/time.

from datetime import datetime, timedelta
from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()

# Run once at specific datetime
scheduler.add_job(
    my_job,
    'date',
    run_date=datetime(2025, 12, 31, 23, 59, 59)
)

# Run once in 30 seconds
scheduler.add_job(
    my_job,
    'date',
    run_date=datetime.now() + timedelta(seconds=30)
)

# Run immediately
scheduler.add_job(my_job, 'date')

# Using string format
scheduler.add_job(
    my_job,
    'date',
    run_date='2025-12-31 23:59:59'
)

Interval Trigger

Runs repeatedly at fixed intervals.

from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime

scheduler = BackgroundScheduler()

# Every 30 seconds
scheduler.add_job(my_job, 'interval', seconds=30)

# Every 5 minutes
scheduler.add_job(my_job, 'interval', minutes=5)

# Every 2 hours
scheduler.add_job(my_job, 'interval', hours=2)

# Every day
scheduler.add_job(my_job, 'interval', days=1)

# Combined intervals
scheduler.add_job(
    my_job,
    'interval',
    hours=2,
    minutes=30,
    seconds=15
)

# With start and end dates
scheduler.add_job(
    my_job,
    'interval',
    minutes=10,
    start_date='2025-01-01 00:00:00',
    end_date='2025-12-31 23:59:59'
)

# Jitter to prevent thundering herd
scheduler.add_job(
    my_job,
    'interval',
    seconds=30,
    jitter=5  # Random delay up to 5 seconds
)

Cron Trigger

Runs based on cron-like expressions for complex schedules.

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()

# Every day at 3am
scheduler.add_job(my_job, 'cron', hour=3)

# Every Monday at 9:30am
scheduler.add_job(my_job, 'cron', day_of_week='mon', hour=9, minute=30)

# Every weekday at 8am
scheduler.add_job(
    my_job,
    'cron',
    day_of_week='mon-fri',
    hour=8
)

# First day of every month at midnight
scheduler.add_job(my_job, 'cron', day=1, hour=0, minute=0)

# Every 15 minutes
scheduler.add_job(my_job, 'cron', minute='*/15')

# At 0, 15, 30, 45 minutes past each hour
scheduler.add_job(my_job, 'cron', minute='0,15,30,45')

# Complex schedule: Mon-Fri at 9am, 12pm, 3pm
scheduler.add_job(
    my_job,
    'cron',
    day_of_week='mon-fri',
    hour='9,12,15',
    minute=0
)

# Last Friday of every month at 5pm
scheduler.add_job(
    my_job,
    'cron',
    day='last fri',
    hour=17
)

# With timezone
scheduler.add_job(
    my_job,
    'cron',
    hour=9,
    timezone='Europe/London'
)

Cron Expression Fields

Field	Values	Special Characters
year	4-digit year	*
month	1-12	* / , -
day	1-31	* / , - last
week	1-53	* / , -
day_of_week	mon-sun or 0-6	* / , -
hour	0-23	* / , -
minute	0-59	* / , -
second	0-59	* / , -

Creating and Managing Jobs

Jobs can be added, modified, paused, resumed, and removed dynamically.

Adding Jobs

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()

# Basic job
scheduler.add_job(my_job, 'interval', seconds=30)

# Job with ID (for later reference)
scheduler.add_job(
    my_job,
    'interval',
    seconds=30,
    id='my_job_id'
)

# Job with arguments
def greet(name, greeting='Hello'):
    print(f"{greeting}, {name}!")

scheduler.add_job(
    greet,
    'interval',
    seconds=30,
    args=['Alice'],
    kwargs={'greeting': 'Hi'}
)

# Job with metadata
scheduler.add_job(
    my_job,
    'interval',
    seconds=30,
    id='report_job',
    name='Daily Report Generator',
    replace_existing=True
)

# Decorator syntax
@scheduler.scheduled_job('interval', seconds=30)
def scheduled_task():
    print("Running scheduled task")

# Immediate first run then interval
scheduler.add_job(
    my_job,
    'interval',
    seconds=30,
    next_run_time=datetime.now()
)

Managing Jobs

# Get a job by ID
job = scheduler.get_job('my_job_id')

# Get all jobs
jobs = scheduler.get_jobs()

# Modify a job
scheduler.modify_job(
    'my_job_id',
    args=['new_arg'],
    name='Updated Job Name'
)

# Reschedule a job (change trigger)
scheduler.reschedule_job(
    'my_job_id',
    trigger='cron',
    hour=3
)

# Pause a job
scheduler.pause_job('my_job_id')

# Resume a job
scheduler.resume_job('my_job_id')

# Remove a job
scheduler.remove_job('my_job_id')

# Remove all jobs
scheduler.remove_all_jobs()

# Check if job exists
if scheduler.get_job('my_job_id'):
    print("Job exists")

Job Configuration Options

scheduler.add_job(
    my_job,
    'interval',
    seconds=30,

    # Identification
    id='unique_job_id',
    name='Human-readable name',

    # Execution control
    max_instances=3,           # Max concurrent executions
    coalesce=True,             # Combine missed runs into one
    misfire_grace_time=60,     # Seconds to allow late execution

    # Replacement behaviour
    replace_existing=True,     # Replace if ID exists

    # Immediate execution
    next_run_time=datetime.now(),

    # Function arguments
    args=['arg1', 'arg2'],
    kwargs={'key': 'value'}
)

Executors

Executors determine how jobs are run (threads, processes, or async).

Thread Pool Executor

Default executor, suitable for I/O-bound tasks.

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.executors.pool import ThreadPoolExecutor

executors = {
    'default': ThreadPoolExecutor(max_workers=20),
    'io_tasks': ThreadPoolExecutor(max_workers=10),
}

scheduler = BackgroundScheduler(executors=executors)

# Assign job to specific executor
scheduler.add_job(
    io_bound_task,
    'interval',
    seconds=30,
    executor='io_tasks'
)

Process Pool Executor

For CPU-bound tasks, runs jobs in separate processes.

from apscheduler.executors.pool import ProcessPoolExecutor

executors = {
    'default': ThreadPoolExecutor(max_workers=10),
    'cpu_tasks': ProcessPoolExecutor(max_workers=4),
}

scheduler = BackgroundScheduler(executors=executors)

# CPU-bound job
scheduler.add_job(
    cpu_intensive_task,
    'interval',
    minutes=5,
    executor='cpu_tasks'
)

AsyncIO Executor

For async functions in asyncio applications.

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.executors.asyncio import AsyncIOExecutor

executors = {
    'default': AsyncIOExecutor(),
}

scheduler = AsyncIOScheduler(executors=executors)

async def async_job():
    await some_async_operation()
    print("Async job completed")

scheduler.add_job(async_job, 'interval', seconds=30)

Multiple Executors Configuration

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor

executors = {
    'default': ThreadPoolExecutor(20),
    'processpool': ProcessPoolExecutor(5)
}

job_defaults = {
    'coalesce': False,
    'max_instances': 3
}

scheduler = BackgroundScheduler(
    executors=executors,
    job_defaults=job_defaults,
    timezone='Europe/London'
)

Job Stores

Job stores persist job data, enabling jobs to survive scheduler restarts.

Memory Job Store

Default store, jobs lost on restart.

from apscheduler.jobstores.memory import MemoryJobStore

jobstores = {
    'default': MemoryJobStore()
}

scheduler = BackgroundScheduler(jobstores=jobstores)

SQLAlchemy Job Store

Persists jobs to any SQLAlchemy-supported database.

from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore

# SQLite
jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}

# PostgreSQL
jobstores = {
    'default': SQLAlchemyJobStore(
        url='postgresql://user:pass@localhost/dbname'
    )
}

# MySQL
jobstores = {
    'default': SQLAlchemyJobStore(
        url='mysql+pymysql://user:pass@localhost/dbname'
    )
}

# With existing engine
from sqlalchemy import create_engine
engine = create_engine('postgresql://user:pass@localhost/dbname')
jobstores = {
    'default': SQLAlchemyJobStore(engine=engine)
}

scheduler = BackgroundScheduler(jobstores=jobstores)

Redis Job Store

Fast, distributed job storage.

from apscheduler.jobstores.redis import RedisJobStore

# Basic configuration
jobstores = {
    'default': RedisJobStore(
        host='localhost',
        port=6379,
        db=0
    )
}

# With authentication
jobstores = {
    'default': RedisJobStore(
        host='localhost',
        port=6379,
        db=0,
        password='secret'
    )
}

# With Redis URL
import redis
redis_client = redis.from_url('redis://localhost:6379/0')
jobstores = {
    'default': RedisJobStore(
        jobs_key='apscheduler.jobs',
        run_times_key='apscheduler.run_times',
        redis=redis_client
    )
}

scheduler = BackgroundScheduler(jobstores=jobstores)

MongoDB Job Store

Document-based job storage.

from apscheduler.jobstores.mongodb import MongoDBJobStore

# Basic configuration
jobstores = {
    'default': MongoDBJobStore(
        host='localhost',
        port=27017,
        database='apscheduler',
        collection='jobs'
    )
}

# With authentication
jobstores = {
    'default': MongoDBJobStore(
        host='localhost',
        port=27017,
        database='apscheduler',
        collection='jobs',
        username='user',
        password='pass',
        authSource='admin'
    )
}

# With existing client
from pymongo import MongoClient
client = MongoClient('mongodb://localhost:27017/')
jobstores = {
    'default': MongoDBJobStore(client=client, database='apscheduler')
}

scheduler = BackgroundScheduler(jobstores=jobstores)

Multiple Job Stores

jobstores = {
    'default': MemoryJobStore(),
    'persistent': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite'),
    'redis': RedisJobStore(host='localhost', port=6379)
}

scheduler = BackgroundScheduler(jobstores=jobstores)

# Add job to specific store
scheduler.add_job(
    my_job,
    'interval',
    seconds=30,
    jobstore='persistent'
)

Error Handling and Job Listeners

Handle job events and errors using listeners.

Event Listeners

from apscheduler.events import (
    EVENT_JOB_EXECUTED,
    EVENT_JOB_ERROR,
    EVENT_JOB_MISSED,
    EVENT_JOB_ADDED,
    EVENT_JOB_REMOVED,
    EVENT_JOB_MODIFIED,
    EVENT_SCHEDULER_STARTED,
    EVENT_SCHEDULER_SHUTDOWN
)

def job_listener(event):
    if event.exception:
        print(f"Job {event.job_id} failed: {event.exception}")
    else:
        print(f"Job {event.job_id} executed successfully")

def error_listener(event):
    print(f"Job {event.job_id} raised {event.exception.__class__.__name__}")
    print(f"Traceback: {event.traceback}")

def missed_listener(event):
    print(f"Job {event.job_id} missed at {event.scheduled_run_time}")

# Add listeners
scheduler.add_listener(
    job_listener,
    EVENT_JOB_EXECUTED | EVENT_JOB_ERROR
)

scheduler.add_listener(error_listener, EVENT_JOB_ERROR)
scheduler.add_listener(missed_listener, EVENT_JOB_MISSED)

Comprehensive Event Handling

from apscheduler.events import (
    EVENT_ALL,
    EVENT_SCHEDULER_STARTED,
    EVENT_SCHEDULER_SHUTDOWN,
    EVENT_SCHEDULER_PAUSED,
    EVENT_SCHEDULER_RESUMED,
    EVENT_JOB_ADDED,
    EVENT_JOB_REMOVED,
    EVENT_JOB_MODIFIED,
    EVENT_JOB_EXECUTED,
    EVENT_JOB_ERROR,
    EVENT_JOB_MISSED,
    EVENT_JOB_SUBMITTED,
    EVENT_JOB_MAX_INSTANCES
)
import logging

logger = logging.getLogger(__name__)

def comprehensive_listener(event):
    if event.code == EVENT_SCHEDULER_STARTED:
        logger.info("Scheduler started")
    elif event.code == EVENT_SCHEDULER_SHUTDOWN:
        logger.info("Scheduler shutdown")
    elif event.code == EVENT_JOB_ADDED:
        logger.info(f"Job added: {event.job_id}")
    elif event.code == EVENT_JOB_REMOVED:
        logger.info(f"Job removed: {event.job_id}")
    elif event.code == EVENT_JOB_EXECUTED:
        logger.info(f"Job executed: {event.job_id}")
    elif event.code == EVENT_JOB_ERROR:
        logger.error(
            f"Job error: {event.job_id} - {event.exception}",
            exc_info=event.traceback
        )
    elif event.code == EVENT_JOB_MISSED:
        logger.warning(
            f"Job missed: {event.job_id} at {event.scheduled_run_time}"
        )
    elif event.code == EVENT_JOB_MAX_INSTANCES:
        logger.warning(
            f"Job {event.job_id} reached max instances"
        )

scheduler.add_listener(comprehensive_listener, EVENT_ALL)

Exception Handling in Jobs

import logging
from functools import wraps

logger = logging.getLogger(__name__)

# Decorator for job error handling
def job_error_handler(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.error(f"Job {func.__name__} failed: {e}")
            # Optionally re-raise or handle
            raise
    return wrapper

@job_error_handler
def risky_job():
    # Job logic that might fail
    result = external_api_call()
    return result

# Retry logic
def job_with_retry(max_retries=3):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    logger.warning(
                        f"Attempt {attempt + 1}/{max_retries} failed: {e}"
                    )
            logger.error(f"All {max_retries} attempts failed")
            raise last_exception
        return wrapper
    return decorator

@job_with_retry(max_retries=3)
def flaky_job():
    # Job that might need retries
    pass

Integration with Web Frameworks

Flask Integration

from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler
import atexit

app = Flask(__name__)

def scheduled_task():
    with app.app_context():
        # Access Flask context (database, config, etc.)
        print("Running scheduled task")

scheduler = BackgroundScheduler()
scheduler.add_job(
    func=scheduled_task,
    trigger='interval',
    seconds=30,
    id='scheduled_task',
    replace_existing=True
)
scheduler.start()

# Shut down scheduler when app exits
atexit.register(lambda: scheduler.shutdown())

@app.route('/')
def index():
    return "Flask with APScheduler"

@app.route('/jobs')
def list_jobs():
    jobs = scheduler.get_jobs()
    return {
        'jobs': [
            {
                'id': job.id,
                'name': job.name,
                'next_run': str(job.next_run_time)
            }
            for job in jobs
        ]
    }

@app.route('/jobs/<job_id>/pause', methods=['POST'])
def pause_job(job_id):
    scheduler.pause_job(job_id)
    return {'status': 'paused'}

if __name__ == '__main__':
    app.run(debug=True, use_reloader=False)  # Disable reloader!

FastAPI Integration

from fastapi import FastAPI, HTTPException
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from contextlib import asynccontextmanager

# Configure scheduler
jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}

scheduler = AsyncIOScheduler(jobstores=jobstores)

async def scheduled_task():
    print("Running async scheduled task")

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    scheduler.add_job(
        scheduled_task,
        'interval',
        seconds=30,
        id='async_task',
        replace_existing=True
    )
    scheduler.start()
    yield
    # Shutdown
    scheduler.shutdown()

app = FastAPI(lifespan=lifespan)

@app.get("/")
async def root():
    return {"message": "FastAPI with APScheduler"}

@app.get("/jobs")
async def list_jobs():
    jobs = scheduler.get_jobs()
    return {
        "jobs": [
            {
                "id": job.id,
                "name": job.name,
                "next_run": str(job.next_run_time)
            }
            for job in jobs
        ]
    }

@app.post("/jobs/{job_id}/run")
async def run_job_now(job_id: str):
    job = scheduler.get_job(job_id)
    if not job:
        raise HTTPException(status_code=404, detail="Job not found")
    job.modify(next_run_time=datetime.now())
    return {"status": "triggered"}

@app.post("/jobs")
async def add_job(interval_seconds: int, job_id: str):
    scheduler.add_job(
        scheduled_task,
        'interval',
        seconds=interval_seconds,
        id=job_id,
        replace_existing=True
    )
    return {"status": "added", "job_id": job_id}

@app.delete("/jobs/{job_id}")
async def remove_job(job_id: str):
    try:
        scheduler.remove_job(job_id)
        return {"status": "removed"}
    except Exception:
        raise HTTPException(status_code=404, detail="Job not found")

Django Integration

# scheduler.py
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from django.conf import settings

def get_scheduler():
    jobstores = {
        'default': SQLAlchemyJobStore(
            url=settings.DATABASE_URL
        )
    }

    scheduler = BackgroundScheduler(jobstores=jobstores)
    return scheduler

scheduler = get_scheduler()

# apps.py
from django.apps import AppConfig

class MyAppConfig(AppConfig):
    name = 'myapp'

    def ready(self):
        from .scheduler import scheduler
        from .tasks import my_task

        scheduler.add_job(
            my_task,
            'interval',
            minutes=30,
            id='my_task',
            replace_existing=True
        )
        scheduler.start()

Clustering and High Availability

Run schedulers across multiple instances whilst avoiding duplicate job execution.

Database-Based Locking

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor

# All instances share the same database
jobstores = {
    'default': SQLAlchemyJobStore(
        url='postgresql://user:pass@localhost/scheduler_db'
    )
}

executors = {
    'default': ThreadPoolExecutor(20)
}

job_defaults = {
    'coalesce': True,
    'max_instances': 1,
    'misfire_grace_time': 60
}

scheduler = BackgroundScheduler(
    jobstores=jobstores,
    executors=executors,
    job_defaults=job_defaults
)

HA caveat: a shared SQLAlchemy (or Redis) jobstore does not prevent duplicate execution — multiple scheduler processes can each fire the same job independently. For true HA, run a single scheduler instance and handle failover externally (e.g. a supervisor, Kubernetes leader-election, or a SETNX-style lock around job logic). APScheduler v4's data-store model is designed to address this properly, but v4 remains pre-release as of mid-2026.

Redis-Based Distributed Locking

import redis
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.redis import RedisJobStore

# Shared Redis for all scheduler instances
redis_client = redis.Redis(host='redis-host', port=6379, db=0)

jobstores = {
    'default': RedisJobStore(redis=redis_client)
}

job_defaults = {
    'coalesce': True,
    'max_instances': 1
}

scheduler = BackgroundScheduler(
    jobstores=jobstores,
    job_defaults=job_defaults
)

# Custom distributed lock wrapper
class DistributedLock:
    def __init__(self, redis_client, lock_name, timeout=60):
        self.redis = redis_client
        self.lock_name = lock_name
        self.timeout = timeout
        self.lock = None

    def __enter__(self):
        self.lock = self.redis.lock(
            self.lock_name,
            timeout=self.timeout
        )
        acquired = self.lock.acquire(blocking=False)
        if not acquired:
            raise RuntimeError("Could not acquire lock")
        return self

    def __exit__(self, *args):
        if self.lock:
            self.lock.release()

def distributed_job():
    try:
        with DistributedLock(redis_client, 'job_lock'):
            # Only one instance executes
            perform_actual_work()
    except RuntimeError:
        # Another instance has the lock
        pass

Leader Election Pattern

import socket
import redis
from datetime import datetime, timedelta

class LeaderElection:
    def __init__(self, redis_client, service_name, ttl=30):
        self.redis = redis_client
        self.service_name = service_name
        self.ttl = ttl
        self.instance_id = f"{socket.gethostname()}-{os.getpid()}"

    def is_leader(self):
        key = f"leader:{self.service_name}"

        # Try to become leader
        acquired = self.redis.set(
            key,
            self.instance_id,
            nx=True,  # Only if not exists
            ex=self.ttl
        )

        if acquired:
            return True

        # Check if we're already leader
        current_leader = self.redis.get(key)
        if current_leader and current_leader.decode() == self.instance_id:
            # Refresh TTL
            self.redis.expire(key, self.ttl)
            return True

        return False

# Use in scheduler
leader = LeaderElection(redis_client, 'my-scheduler')

def leader_only_job():
    if not leader.is_leader():
        return  # Skip if not leader

    # Perform the actual job
    perform_work()

scheduler.add_job(
    leader_only_job,
    'interval',
    seconds=30
)

Common Use Cases

Periodic Data Cleanup

from datetime import datetime, timedelta

def cleanup_old_records():
    cutoff = datetime.now() - timedelta(days=30)
    deleted = db.session.query(TempData).filter(
        TempData.created_at < cutoff
    ).delete()
    db.session.commit()
    print(f"Deleted {deleted} old records")

scheduler.add_job(
    cleanup_old_records,
    'cron',
    hour=2,  # Run at 2am daily
    id='cleanup_job'
)

Report Generation

def generate_daily_report():
    data = fetch_daily_metrics()
    report = create_report(data)
    send_email(
        to='team@example.com',
        subject=f"Daily Report - {datetime.now().date()}",
        body=report
    )

scheduler.add_job(
    generate_daily_report,
    'cron',
    day_of_week='mon-fri',
    hour=8,
    minute=0,
    id='daily_report'
)

API Data Synchronisation

def sync_external_data():
    response = requests.get('https://api.example.com/data')
    data = response.json()

    for item in data['items']:
        update_or_create(item)

    print(f"Synced {len(data['items'])} items")

scheduler.add_job(
    sync_external_data,
    'interval',
    minutes=15,
    id='data_sync',
    max_instances=1  # Prevent overlapping
)

Health Check Monitoring

import requests

def health_check():
    services = [
        ('API', 'https://api.example.com/health'),
        ('Database', 'https://db.example.com/health'),
        ('Cache', 'https://cache.example.com/health'),
    ]

    for name, url in services:
        try:
            response = requests.get(url, timeout=5)
            if response.status_code != 200:
                alert(f"{name} unhealthy: {response.status_code}")
        except requests.RequestException as e:
            alert(f"{name} unreachable: {e}")

scheduler.add_job(
    health_check,
    'interval',
    minutes=1,
    id='health_check'
)

Cache Warming

def warm_cache():
    popular_items = get_popular_item_ids()

    for item_id in popular_items:
        data = fetch_item_from_db(item_id)
        cache.set(f"item:{item_id}", data, timeout=3600)

    print(f"Warmed cache for {len(popular_items)} items")

scheduler.add_job(
    warm_cache,
    'cron',
    hour='*/4',  # Every 4 hours
    id='cache_warm'
)

Performance Considerations

Executor Tuning

from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor

# Size thread pool based on workload
# I/O-bound: Higher count (20-100)
# CPU-bound: Use ProcessPoolExecutor with CPU count

executors = {
    'default': ThreadPoolExecutor(20),        # I/O tasks
    'cpu': ProcessPoolExecutor(4),            # CPU tasks
    'limited': ThreadPoolExecutor(5),         # Rate-limited tasks
}

# Assign jobs to appropriate executor
scheduler.add_job(io_task, executor='default')
scheduler.add_job(cpu_task, executor='cpu')
scheduler.add_job(api_call, executor='limited')

Job Store Optimisation

# SQLAlchemy with connection pooling
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    'postgresql://user:pass@localhost/db',
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20
)

jobstores = {
    'default': SQLAlchemyJobStore(engine=engine)
}

# Redis with connection pooling
import redis
pool = redis.ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    max_connections=20
)
redis_client = redis.Redis(connection_pool=pool)

jobstores = {
    'default': RedisJobStore(redis=redis_client)
}

Preventing Job Pile-up

job_defaults = {
    'coalesce': True,           # Combine missed runs
    'max_instances': 1,         # Only one instance at a time
    'misfire_grace_time': 60    # Allow 60s grace period
}

scheduler = BackgroundScheduler(job_defaults=job_defaults)

# Per-job settings
scheduler.add_job(
    slow_job,
    'interval',
    seconds=30,
    max_instances=1,      # Prevent overlap
    coalesce=True,        # Skip missed runs
    misfire_grace_time=10 # Short grace period
)

Memory Management

# Use generators for large data processing
def process_large_dataset():
    for batch in fetch_batches(size=1000):
        process_batch(batch)
        # Memory freed after each batch

# Avoid storing large results in job
def efficient_job():
    # Process and save incrementally
    for item in items:
        result = process(item)
        save_result(result)
    # Don't return large data

Quick Reference

Scheduler Methods

Method	Description
`start()`	Start the scheduler
`shutdown(wait=True)`	Stop the scheduler
`pause()`	Pause job processing
`resume()`	Resume job processing
`add_job()`	Add a new job
`remove_job(job_id)`	Remove a job
`get_job(job_id)`	Get job by ID
`get_jobs()`	Get all jobs
`modify_job(job_id, **changes)`	Modify job properties
`reschedule_job(job_id, trigger, **trigger_args)`	Change job trigger
`pause_job(job_id)`	Pause a specific job
`resume_job(job_id)`	Resume a paused job
`print_jobs()`	Print all jobs to stdout

Trigger Quick Reference

# Date - run once
'date', run_date='2025-12-31 23:59:59'
'date', run_date=datetime(2025, 12, 31)

# Interval - repeat fixed interval
'interval', seconds=30
'interval', minutes=5
'interval', hours=2
'interval', days=1

# Cron - complex schedules
'cron', hour=3                    # Daily at 3am
'cron', hour=9, minute=30         # Daily at 9:30am
'cron', day_of_week='mon-fri'     # Weekdays
'cron', minute='*/15'             # Every 15 minutes
'cron', day=1, hour=0             # First of month

Common Imports

# Schedulers
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.schedulers.asyncio import AsyncIOScheduler

# Job stores
from apscheduler.jobstores.memory import MemoryJobStore
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.jobstores.redis import RedisJobStore
from apscheduler.jobstores.mongodb import MongoDBJobStore

# Executors
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
from apscheduler.executors.asyncio import AsyncIOExecutor

# Events
from apscheduler.events import (
    EVENT_JOB_EXECUTED,
    EVENT_JOB_ERROR,
    EVENT_JOB_MISSED
)

CLI Commands

# Install APScheduler
uv add apscheduler

# With specific backends
uv add apscheduler[sqlalchemy]
uv add apscheduler[redis]
uv add apscheduler[mongodb]

# All backends
uv add apscheduler[sqlalchemy,redis,mongodb]

Common Issues and Solutions

Issue	Solution
Jobs run twice with Flask debug mode	Use `use_reloader=False` in `app.run()` or check `WERKZEUG_RUN_MAIN` env var
Jobs not persisting after restart	Use a persistent job store (SQLAlchemy, Redis, MongoDB) instead of MemoryJobStore
Jobs missed during downtime	Set `coalesce=True` and appropriate `misfire_grace_time`
"Maximum instances reached" warning	Increase `max_instances` or ensure jobs complete faster
Jobs overlap/run concurrently	Set `max_instances=1` for the job
Timezone issues	Set timezone explicitly: `scheduler = BackgroundScheduler(timezone='Europe/London')`
Job not found after restart	Ensure job uses `replace_existing=True` and persistent job store
Database connection errors	Use connection pooling; handle reconnection in job functions
Memory leaks	Avoid storing large data in jobs; use generators for large datasets
"Scheduler already running"	Check for multiple `start()` calls; use singleton pattern

Debugging Tips

# Enable logging
import logging
logging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.DEBUG)

# Print all scheduled jobs
scheduler.print_jobs()

# Check next run time
job = scheduler.get_job('my_job')
print(f"Next run: {job.next_run_time}")

# Verify scheduler state
print(f"Running: {scheduler.running}")
print(f"State: {scheduler.state}")

# List job store contents
for job in scheduler.get_jobs():
    print(f"{job.id}: {job.next_run_time} - {job.func}")

Flask Debug Mode Fix

import os
from flask import Flask
from apscheduler.schedulers.background import BackgroundScheduler

app = Flask(__name__)

# Only start scheduler in main process
if os.environ.get('WERKZEUG_RUN_MAIN') == 'true' or not app.debug:
    scheduler = BackgroundScheduler()
    scheduler.add_job(my_job, 'interval', seconds=30)
    scheduler.start()

if __name__ == '__main__':
    app.run(debug=True)

Graceful Shutdown

import atexit
import signal

scheduler = BackgroundScheduler()
scheduler.start()

def shutdown():
    scheduler.shutdown(wait=False)

# Register shutdown handlers
atexit.register(shutdown)
signal.signal(signal.SIGTERM, lambda *args: shutdown())
signal.signal(signal.SIGINT, lambda *args: shutdown())