How to monitor APScheduler jobs in production

APScheduler is the most widely used scheduling library in the Python ecosystem. It supports cron expressions, interval-based schedules, and one-off delayed execution. It runs cleanly inside a Flask, FastAPI, or Django application, or as a standalone process.

The in-process architecture is also its core monitoring weakness. APScheduler can only observe what happens inside the process it runs in. If the process crashes, the container restarts, or the server reboots, APScheduler stops running entirely. From inside the process, there's no way to know this happened — because the process is gone. From outside, there's no alert unless you've built external monitoring explicitly.

Beyond process failure, APScheduler has additional failure modes that require external visibility: job executor exhaustion (the thread pool is full and jobs are being dropped), missed fire time handling (jobs that fire late due to executor load), and the standard silent failure pattern where a job completes without error but produces no useful output.

APScheduler's monitoring blind spots

Process failure. APScheduler's scheduler lives in the same process as your application. A memory error, an unhandled exception at startup, or a container restart stops all scheduled jobs. APScheduler has no mechanism to detect its own absence.

Executor exhaustion. APScheduler uses thread or process executors to run jobs concurrently. By default, the thread pool executor has a maximum of 10 threads. When all threads are in use and a new job fires, APScheduler logs a warning and drops the job. No external alert fires.

Missed fire time. APScheduler has a misfire_grace_time setting. If a job fires more than misfire_grace_time seconds late (due to executor load or process sleep), APScheduler can be configured to skip it rather than run late. These skips are logged but not alerted.

Silent failures. A job function that returns normally without raising an exception is considered successful by APScheduler, regardless of what it accomplished.

Adding external monitoring to APScheduler jobs

The monitoring pattern wraps each job function with start, success, and fail pings to an external monitoring service. APScheduler's event listeners provide a cleaner integration point than modifying every job function individually.

Option 1: Job-level wrapper (simplest)

import os
import requests
import traceback
from functools import wraps
from typing import Callable, Any

CRONTIFY_API_KEY = os.environ["CRONTIFY_API_KEY"]
BASE_URL = "https://api.crontify.com/api/v1/ping"
HEADERS = {"X-API-Key": CRONTIFY_API_KEY}


def ping(monitor_id: str, event: str, payload: dict | None = None) -> None:
    try:
        requests.post(
            f"{BASE_URL}/{monitor_id}/{event}",
            headers=HEADERS,
            json=payload,
            timeout=5,
        )
    except Exception:
        pass


def monitored(monitor_id: str) -> Callable:
    """Decorator that wraps an APScheduler job with monitoring pings."""
    def decorator(fn: Callable) -> Callable:
        @wraps(fn)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            ping(monitor_id, "start")
            try:
                result = fn(*args, **kwargs)
                meta = result if isinstance(result, dict) else {}
                ping(monitor_id, "success",
                     {"meta": meta} if meta else None)
                return result
            except Exception as exc:
                ping(monitor_id, "fail", {
                    "message": str(exc),
                    "log": traceback.format_exc(),
                })
                raise
        return wrapper
    return decorator

Apply it to any APScheduler job function:

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()


@monitored("mon_abc123")
def nightly_sync() -> dict:
    result = sync_records()
    return {
        "records_synced": result.count,
        "duration_ms": result.duration_ms,
    }


@monitored("mon_def456")
def daily_report() -> dict:
    report = generate_report()
    return {
        "rows_in_report": report.row_count,
        "recipients_emailed": report.recipients,
    }


scheduler.add_job(nightly_sync, "cron", hour=2, minute=0)
scheduler.add_job(daily_report, "cron", hour=6, minute=0)
scheduler.start()

The @monitored decorator sends a start ping before the function runs and a success ping with the returned dict as metadata when it completes. Exceptions trigger a fail ping with the full traceback.

Option 2: APScheduler event listeners (scheduler-level)

APScheduler fires events for job execution, errors, and missed fires. You can attach listeners to send pings without modifying individual job functions:

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

MONITOR_MAP = {
    "nightly_sync": "mon_abc123",
    "daily_report": "mon_def456",
}


def on_job_executed(event):
    monitor_id = MONITOR_MAP.get(event.job_id)
    if not monitor_id:
        return
    ping(monitor_id, "success")


def on_job_error(event):
    monitor_id = MONITOR_MAP.get(event.job_id)
    if not monitor_id:
        return
    ping(monitor_id, "fail", {
        "message": str(event.exception),
        "log": "".join(traceback.format_tb(event.traceback)),
    })


def on_job_missed(event):
    monitor_id = MONITOR_MAP.get(event.job_id)
    if not monitor_id:
        return
    ping(monitor_id, "fail", {
        "message": f"Job {event.job_id} missed its scheduled execution",
    })


scheduler.add_listener(on_job_executed, EVENT_JOB_EXECUTED)
scheduler.add_listener(on_job_error, EVENT_JOB_ERROR)
scheduler.add_listener(on_job_missed, EVENT_JOB_MISSED)

The event listener approach is cleaner for adding monitoring to existing APScheduler setups with many jobs — you add listeners once and every job in MONITOR_MAP gets coverage without touching the job functions.

The limitation: event listeners can't easily attach job output metadata to the success ping because they don't have access to the job's return value by default. For metadata-based alert rules, the decorator approach (Option 1) is better.

Detecting process failure

APScheduler event listeners only fire when APScheduler is running. If the process dies, no event fires and no ping is sent. External monitoring detects this because the monitoring service knows when the ping was expected.

For the external monitoring to catch process failures, it needs a start ping to know the job was supposed to run. You can also add a process-level heartbeat — a separate job that pings every minute to confirm the scheduler is alive:

def scheduler_heartbeat() -> None:
    """Fired every minute to confirm the APScheduler process is running."""
    ping("mon_heartbeat_123", "success")


scheduler.add_job(scheduler_heartbeat, "interval", minutes=1)

Configure the corresponding monitor in Crontify with a 2-minute grace period. If the scheduler process stops, no heartbeat arrives, and an alert fires within 2 minutes.

Silent failure detection

The dict returned from your job function in Option 1 is sent as metadata on the success ping. Define alert rules in Crontify's dashboard:

recipients_emailed eq 0 → fire alert
rows_in_report eq 0 → fire alert

An APScheduler job that completes without raising an exception but produces no output will trigger an alert. This is the failure class APScheduler — and most monitoring tools — was never designed to catch.

Frequently asked questions

Does this work with AsyncIOScheduler?

Yes, with minor adjustments. Replace the synchronous requests.post in the ping helper with aiohttp or httpx and make the helper an async function. The @monitored decorator needs to handle both sync and async job functions:

import asyncio

def monitored(monitor_id: str) -> Callable:
    def decorator(fn: Callable) -> Callable:
        @wraps(fn)
        async def async_wrapper(*args, **kwargs):
            ping(monitor_id, "start")
            try:
                result = await fn(*args, **kwargs)
                meta = result if isinstance(result, dict) else {}
                ping(monitor_id, "success", {"meta": meta} if meta else None)
                return result
            except Exception as exc:
                ping(monitor_id, "fail", {
                    "message": str(exc),
                    "log": traceback.format_exc(),
                })
                raise

        @wraps(fn)
        def sync_wrapper(*args, **kwargs):
            ping(monitor_id, "start")
            try:
                result = fn(*args, **kwargs)
                meta = result if isinstance(result, dict) else {}
                ping(monitor_id, "success", {"meta": meta} if meta else None)
                return result
            except Exception as exc:
                ping(monitor_id, "fail", {
                    "message": str(exc),
                    "log": traceback.format_exc(),
                })
                raise

        return async_wrapper if asyncio.iscoroutinefunction(fn) else sync_wrapper
    return decorator

What misfire_grace_time should I set?

Set it to slightly longer than the maximum acceptable delay for each job. For a job that must run within 5 minutes of its scheduled time, use misfire_grace_time=300. For jobs where the exact time matters less, a longer grace time reduces false missed-fire events.

Crontify is free for up to 5 monitors — no credit card required.