← Back to DevBytes
Python July 11, 2026 9 min read

How to Create a Cron Job Scheduler with Python

Introduction to Cron Job Scheduling in Python

A cron job scheduler automates the execution of tasks at predefined times or intervals. The name comes from the Unix utility cron, which reads a configuration file (crontab) and runs commands according to a schedule. In Python, we can build our own scheduling logic or leverage robust libraries that mimic cron's expression syntax while adding features like persistence, error handling, and dynamic job management.

This tutorial walks you through creating a cron job scheduler entirely in Python. You’ll learn the core concepts, implement a basic scheduler from scratch, then move on to using production-ready libraries like schedule and APScheduler. By the end, you’ll be able to design a scheduler that fits your application’s needs—whether it’s a simple script that sends emails every Monday or a complex system with thousands of jobs stored in a database.

Why Build a Custom Python Scheduler?

🚀 Deploy your AI agent in 10 minutes

Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.

Try it free →

You might ask: why not just use the operating system’s cron? There are several reasons to embed scheduling directly inside a Python application:

Core Concepts and Terminology

Before writing code, let's establish the key building blocks of any cron-like scheduler:

Job / Task

A callable (function, method, or any object implementing __call__) that performs the actual work. It could be a simple print statement, a database cleanup, or an API call.

Trigger / Schedule

The rule that determines when a job runs. Triggers can be time-based (e.g., every 5 minutes, daily at 02:00) or event-based. Cron-style triggers use a string of five fields (minute hour day month day_of_week) to express complex schedules like “every weekday at 08:30”.

Scheduler / Engine

The component that checks triggers, launches jobs when they are due, and manages concurrency. It often runs on a background thread or process and maintains an internal job store.

Job Store

Where job definitions and their next run times are kept. For simple in-memory schedulers this is a list or dictionary. For production, jobs may be persisted to a database so they survive application restarts.

Executor

Determines how a job is actually executed—synchronously in the scheduler’s thread, in a thread pool, or in a separate process. This affects concurrency and isolation.

Getting Started: A Minimal Scheduler from Scratch

Let’s build a tiny scheduler to understand the mechanics. It runs a function every n seconds. This example is intentionally simple and not suitable for production, but it shows the loop-and-check pattern. 

import time
import threading
from datetime import datetime, timedelta

class SimpleScheduler:
    def __init__(self):
        self.jobs = []

    def every(self, interval_seconds, func, *args, **kwargs):
        """Schedule a function to run every `interval_seconds` seconds."""
        job = {
            'interval': timedelta(seconds=interval_seconds),
            'next_run': datetime.now() + timedelta(seconds=interval_seconds),
            'func': func,
            'args': args,
            'kwargs': kwargs
        }
        self.jobs.append(job)
        return job

    def run(self):
        """Main loop: check jobs and execute when due."""
        while True:
            now = datetime.now()
            for job in self.jobs:
                if now >= job['next_run']:
                    # Execute the job (in a thread for non‑blocking behavior)
                    threading.Thread(
                        target=job['func'],
                        args=job['args'],
                        kwargs=job['kwargs']
                    ).start()
                    # Schedule next run
                    job['next_run'] = now + job['interval']
            time.sleep(1)  # Check resolution of 1 second

# Example usage
def greet(name):
    print(f"{datetime.now().strftime('%H:%M:%S')} - Hello, {name}!")

scheduler = SimpleScheduler()
scheduler.every(5, greet, "Alice")
scheduler.every(10, greet, "Bob")

# Run in a background thread so the main program can do other things
threading.Thread(target=scheduler.run, daemon=True).start()

# Keep the main thread alive to observe output
try:
    while True:
        time.sleep(30)
except KeyboardInterrupt:
    print("Scheduler stopped.")

This illustrates the core loop: iterate over jobs, compare the current time against each job’s next_run, and launch the function when overdue. However, it lacks cron‑style expression support, persistence, and proper error handling. For real projects, we turn to mature libraries.

Using the schedule Library for Simple Jobs

The schedule library is a pure-Python scheduler with a friendly, fluent API. It’s ideal for scripts and lightweight applications. Install it with pip install schedule.

Basic Scheduling with schedule

You express schedules using natural methods like every().day.at("08:00") or every(10).minutes.do(job). The main loop is a simple schedule.run_pending() call inside a while loop.

import schedule
import time
from datetime import datetime

def send_report():
    print(f"[{datetime.now().strftime('%H:%M:%S')}] Generating daily report...")

def backup_database():
    print(f"[{datetime.now().strftime('%H:%M:%S')}] Starting database backup...")

# Schedule tasks
schedule.every().day.at("06:30").do(send_report)
schedule.every().monday.at("02:00").do(backup_database)

# A task that runs every 15 minutes
schedule.every(15).minutes.do(lambda: print("15-minute heartbeat"))

print("Scheduler started. Press Ctrl+C to exit.")
while True:
    schedule.run_pending()
    time.sleep(1)

Adding Cron‑style Expressions

The schedule library doesn't natively parse standard cron expressions, but you can simulate most patterns with its chaining methods. For full cron syntax, consider APScheduler (next section). However, you can combine rules creatively: 

# Every weekday (Monday through Friday) at 09:00
schedule.every().monday.at("09:00").do(meeting_reminder)
schedule.every().tuesday.at("09:00").do(meeting_reminder)
schedule.every().wednesday.at("09:00").do(meeting_reminder)
schedule.every().thursday.at("09:00").do(meeting_reminder)
schedule.every().friday.at("09:00").do(meeting_reminder)

# Alternatively, use a conditional inside a generic job
def weekday_morning_job():
    if datetime.today().weekday() < 5:  # Monday=0 to Friday=4
        meeting_reminder()

schedule.every().day.at("09:00").do(weekday_morning_job)

Error Handling and Logging

By default, schedule silently catches exceptions thrown by jobs. You can inspect failed jobs via schedule.jobs and their .exception attribute. For better visibility, wrap your job functions with try/except and logging. 

import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')

def robust_task():
    try:
        # actual work
        pass
    except Exception as e:
        logging.error(f"Task failed: {e}")

schedule.every(5).minutes.do(robust_task)

Advanced Scheduling with APScheduler

APScheduler (Advanced Python Scheduler) is the go‑to library for production-grade scheduling. It supports:

Install it with pip install apscheduler.

A Basic APScheduler with Cron Trigger

Here we set up a background scheduler using a cron trigger that mimics a classic crontab entry. The job runs at 08:00 every weekday.

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
import logging
import time

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def weekday_morning_task():
    logger.info("Running weekday morning task")

# Create a background scheduler
scheduler = BackgroundScheduler()

# Add a job with a cron trigger: minute=0, hour=8, day_of_week='mon-fri'
scheduler.add_job(
    weekday_morning_task,
    trigger=CronTrigger(
        minute=0,
        hour=8,
        day_of_week='mon-fri'
    ),
    id='morning_job',
    name='Weekday morning task',
    replace_existing=True
)

scheduler.start()

try:
    # Keep the main thread alive
    while True:
        time.sleep(2)
except KeyboardInterrupt:
    scheduler.shutdown()

Using a Cron Expression String

APScheduler also accepts a standard cron expression via the cron trigger’s cron parameter. The string must contain five fields separated by spaces (minute, hour, day, month, day_of_week).

# "At 02:30 every day" → '30 2 * * *'
scheduler.add_job(
    my_daily_job,
    trigger='cron',
    cron='30 2 * * *',
    id='daily_backup'
)

Persisting Jobs in a Database

To survive restarts, configure APScheduler with a SQLAlchemy job store. First install sqlalchemy and the appropriate database driver. Then initialize the scheduler with a job store pointing to your database URL. 

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

jobstores = {
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
executors = {
    'default': ThreadPoolExecutor(20)
}
scheduler = BackgroundScheduler(
    jobstores=jobstores,
    executors=executors
)

# Add a job – it will be stored in the SQLite database
def my_persistent_job():
    print("Job executed")

scheduler.add_job(
    my_persistent_job,
    trigger='interval',
    seconds=10,
    id='persistent_job'
)

scheduler.start()

After restarting the application, APScheduler will reload the job definitions from the database and resume scheduling. This is critical for long‑running services.

Managing Jobs Dynamically

APScheduler allows you to inspect, pause, resume, and remove jobs by their ID.

# List all jobs
for job in scheduler.get_jobs():
    print(f"Job {job.id}: next run at {job.next_run_time}")

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

# Resume it later
scheduler.resume_job('morning_job')

# Remove a job permanently
scheduler.remove_job('persistent_job')

# Modify a job's trigger or function
scheduler.reschedule_job(
    'morning_job',
    trigger=CronTrigger(minute=0, hour=9)  # shift from 08:00 to 09:00
)

Integrating with System Cron as a Fallback

Sometimes the simplest solution is to use the operating system’s cron to launch your Python script. You can combine this with an internal scheduler for lightweight task coordination. For example, you could have a cron entry that runs every minute and triggers a Python script that checks a job queue.

A typical crontab line to run a script every 10 minutes:

*/10 * * * * /usr/bin/python3 /path/to/your_script.py

Inside your_script.py, you might use schedule to run multiple tasks in sequence, or you might use a task queue like Celery with scheduled beat support. This approach offloads the “wake‑up” to cron, while Python handles the logic.

Best Practices for Python Cron Schedulers

When building a scheduler that will run in production, keep these guidelines in mind:

Conclusion

Creating a cron job scheduler in Python gives you portable, testable, and deeply integrated task automation. You started by understanding the basic loop and built a minimal scheduler. Then you explored schedule for its simple, fluent API that suits scripts and small applications. For enterprise‑grade needs, APScheduler delivers cron expression support, database persistence, and fine‑grained control over execution and job management. By following best practices around persistence, error handling, and monitoring, you can deploy a scheduler that reliably runs thousands of jobs across any environment. Whether you’re sending periodic emails, cleaning up old data, or orchestrating complex workflows, Python’s scheduling ecosystem has you covered.

🚀 Need a reliable AI agent for your project?

Deploy Hermes Agent in 10 minutes. Managed hosting, zero DevOps.

Get Started — $23.99/mo
← Back to all articles