Stage 7 · Master

Automation Scripts

Scheduling & Cron Alternatives

APScheduler, schedule library, and building scheduled tasks in Python.

6 min readPython for DevOpsMaster

Cron Is Still King

For simple scheduled tasks, cron is the right tool. It is reliable, well-understood, and requires no Python process. Use Python scheduling libraries only when you need in-process scheduling, complex trigger logic, or job persistence.

When to use Python scheduling

Use cron for periodic scripts that run independently. Use Python scheduling when the scheduler must be part of a larger application — a web server, a long-running daemon, or a service that manages its own jobs.

The schedule Library

Simple scheduling with the schedule library


import schedule
import time
from datetime import datetime

def check_service_health():
    print(f"[{datetime.now()}] Checking service health...")

def rotate_logs():
    print(f"[{datetime.now()}] Rotating logs...")

# Schedule jobs
schedule.every(5).minutes.do(check_service_health)
schedule.every().hour.do(rotate_logs)
schedule.every().day.at("02:00).do(cleanup_old_backups)
schedule.every().monday.at("09:00").do(weekly_report)

# Run forever
while True:
    schedule.run_pending()
    time.sleep(1)

schedule is the simplest Python scheduler. It runs in a single thread and keeps the process alive. Good for small scripts that run on a single machine.

APScheduler

APScheduler is the full-featured scheduler for Python. It supports cron-like triggers, job persistence, and multiple execution backends.

APScheduler with cron triggers


from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger

scheduler = BlockingScheduler()

# Interval trigger — every 30 minutes
@scheduler.scheduled_job(IntervalTrigger(minutes=30))
def health_check():
    print("Running health check...")

# Cron trigger — every weekday at 9am
@scheduler.scheduled_job(CronTrigger(day_of_week="mon-fri", hour=9))
def daily_report():
    print("Generating daily report...")

# Cron trigger — first day of every month at midnight
@scheduler.scheduled_job(CronTrigger(day=1, hour=0))
def monthly_cleanup():
    print("Running monthly cleanup...")

scheduler.start()

APScheduler supports three trigger types: interval (fixed intervals), cron (calendar-based), and date (one-shot at a specific time). Choose the one that fits your use case.

systemd Timers

For production servers, systemd timers are more reliable than cron. They support dependency ordering, logging to the journal, and missed-run handling.

systemd timer unit files


# /etc/systemd/system/health-check.service
[Unit]
Description=Service Health Check

[Service]
Type=oneshot
ExecStart=/opt/scripts/health_check.py
WorkingDirectory=/opt/scripts

# /etc/systemd/system/health-check.timer
[Unit]
Description=Run health check every 5 minutes

[Timer]
OnBootSec=1min
OnUnitActiveSec=5min
Persistent=true

[Install]
WantedBy=timers.target

Persistent=true means if the system was off when a timer fired, it will run as soon as the system boots. This is something cron does not do reliably.

Practical Patterns

Scheduled task with locking


import fcntl
import sys
from pathlib import Path
from datetime import datetime

def run_with_lock(lockfile: str, func):
    """Ensure only one instance of the task runs at a time."""
    lock_path = Path(f"/tmp/{lockfile}.lock")
    try:
        fd = open(lock_path, "w")
        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        fd.write(str(__import__("os").getpid()))
        fd.flush()
        func()
    except OSError:
        print(f"[{datetime.now()}] {lockfile} already running, skipping")
    finally:
        try:
            fcntl.flock(fd, fcntl.LOCK_UN)
            fd.close()
            lock_path.unlink(missing_ok=True)
        except Exception:
            pass

# Usage
def daily_backup():
    print("Running backup...")

run_with_lock("daily_backup", daily_backup)

File locking prevents duplicate runs when cron fires before the previous execution finishes. This is critical for tasks that modify shared state.

Error Handling in Scheduled Tasks

Robust scheduled task wrapper


import logging
import traceback
from datetime import datetime

logger = logging.getLogger(__name__)

def scheduled_task(func):
    """Decorator that logs errors and continues running."""
    def wrapper(*args, **kwargs):
        task_name = func.__name__
        try:
            logger.info(f"Starting {task_name}")
            func(*args, **kwargs)
            logger.info(f"Completed {task_name}")
        except Exception as e:
            logger.error(f"Failed {task_name}: {e}")
            logger.debug(traceback.format_exc())
    return wrapper

@scheduled_task
def check_disk_space():
    import shutil
    usage = shutil.disk_usage("/")
    free_gb = usage.free / (1024**3)
    if free_gb < 10:
        logger.warning(f"Low disk space: {free_gb:.1f} GB free")

@scheduled_task
def rotate_logs():
    from pathlib import Path
    for log in Path("/var/log/app").glob("*.log.*.gz"):
        if log.stat().st_mtime < time.time() - 30 * 86400:
            log.unlink()

The decorator catches all exceptions and logs them. The scheduler continues running even if individual tasks fail. This prevents one bad task from stopping all scheduled work.

Always handle exceptions in scheduled tasks

An unhandled exception in a scheduled task can crash the entire scheduler. Wrap every task in try/except and log the error. Never let a scheduled task crash silently.

Mark this lesson complete to store local progress and unlock a cleaner resume path the next time you visit.