Monitoring Python scheduled jobs (scripts, Django commands, APScheduler)

Python jobs usually run in one of three ways: a script started by cron, a Django management command started by cron, or an in-process scheduler like APScheduler. In all three, a heartbeat is the simplest reliable check. The job pings when it's done, and you're alerted when the ping is late or reports a failure.

A dependency-free helper

# heartbeat.py
import contextlib
import logging
import urllib.request


def ping(url: str) -> None:
    try:
        urllib.request.urlopen(url, timeout=5).close()
    except Exception as exc:  # never let monitoring break the job
        logging.warning("heartbeat ping failed: %s", exc)


@contextlib.contextmanager
def monitored(url: str):
    ping(url + "/start")
    try:
        yield
    except BaseException:
        ping(url + "/fail")
        raise
    ping(url)

A plain script run by cron

import os
from heartbeat import monitored

with monitored(os.environ["CLEANUP_PING_URL"]):
    cleanup_old_files()

Or skip code changes entirely and wrap the cron line with cronpup-run: 0 * * * * cronpup-run https://cronpup.com/p/UUID python3 /opt/jobs/cleanup.py

Django management commands

class Command(BaseCommand):
    def handle(self, *args, **options):
        with monitored(settings.NIGHTLY_SYNC_PING_URL):
            sync_everything()

Run it from cron (python manage.py nightly_sync) and give the monitor the same cron expression.

APScheduler

def report_job():
    with monitored(os.environ["REPORT_PING_URL"]):
        build_report()

scheduler.add_job(report_job, "cron", hour=3, minute=0, timezone="Europe/Berlin")

An in-process scheduler stops when the process stops. A heartbeat is how you find out. Add a one-minute heartbeat job too, so you hear about a dead scheduler process within minutes, not at the next nightly run.

For Celery, see monitoring Celery beat. Monitor your Python jobs. 10 monitors are free.