Recurring Job Scheduling with Cron
In this tutorial, you will learn about Recurring Job Scheduling with Cron. We cover key concepts, practical examples, and best practices to help you master this topic.
Implement recurring background job scheduling using cron expressions, interval timers, calendar-based triggers, and database-backed schedulers for production systems.
What You Learn
You will learn how to design recurring job schedules, handle timezone-aware scheduling, prevent overlapping executions, and build a database-backed scheduler for dynamic schedule management.
Why It Matters
Recurring jobs automate critical maintenance like backups, cache warming, and report generation. A robust scheduler guarantees these tasks execute reliably at the right time.
Real-World Use
DodaTech's security platform runs malware signature updates every 2 hours, database vacuum at 3 AM daily, weekly Compliance reports on Monday, and log rotation every 6 hours. Schedules are stored in the database for runtime changes.
Recurring Schedule Types
flowchart TD
S[Scheduler] --> CT{Cron Type}
CT -->|Fixed Interval| I[Every N seconds]
CT -->|Cron Expression| C[Specific times]
CT -->|Calendar| CA[Business hours]
CT -->|Dynamic| D[DB-backed schedule]
I --> W[Worker Pool]
C --> W
CA --> W
D --> W
W --> E[Execute Job]
Cron-Based Recurring Scheduler
import time
from datetime import datetime
import threading
class CronRecurringScheduler:
def __init__(self):
self.jobs = {}
self.running = True
def add_job(self, name, cron_expr, func, timezone='UTC'):
minute, hour, dom, month, dow = cron_expr.split()
self.jobs[name] = {
'func': func,
'cron': {'minute': minute, 'hour': hour, 'dom': dom,
'month': month, 'dow': dow},
'last_run': None,
'timezone': timezone,
}
def _matches(self, cron, now):
def field_matches(pattern, value):
if pattern == '*':
return True
for part in pattern.split(','):
if '/' in part:
base, step = part.split('/')
base_val = int(base) if base != '*' else 0
if (value - base_val) % int(step) == 0:
return True
elif '-' in part:
low, high = part.split('-')
if int(low) <= value <= int(high):
return True
elif pattern.isdigit() or (pattern[0] == '*' and pattern != '*'):
continue
elif part == str(value):
return True
return pattern == str(value) or pattern == '*'
return (field_matches(cron['minute'], now.minute) and
field_matches(cron['hour'], now.hour) and
field_matches(cron['dom'], now.day) and
field_matches(cron['month'], now.month) and
field_matches(cron['dow'], now.weekday()))
def tick(self):
now = datetime.utcnow()
for name, job in self.jobs.items():
if self._matches(job['cron'], now):
if job['last_run'] != f"{now.hour}:{now.minute}":
print(f"Executing recurring: {name}")
job['func']()
job['last_run'] = f"{now.hour}:{now.minute}"
def start(self):
while self.running:
self.tick()
time.sleep(30)
def stop(self):
self.running = False
def backup_db():
print(" Database backup completed")
def warm_cache():
print(" Cache warmed up")
def rotate_logs():
print(" Logs rotated")
sched = CronRecurringScheduler()
sched.add_job('backup', '0 3 * * *', backup_db)
sched.add_job('cache_warm', '*/30 * * * *', warm_cache)
sched.add_job('log_rotate', '0 */6 * * *', rotate_logs)
t = threading.Thread(target=sched.start, daemon=True)
t.start()
time.sleep(35)
sched.stop()
Expected output:
Executing recurring: cache_warm
Executing recurring: cache_warm
Interval-Based Recurring Scheduler
import time
import threading
class IntervalRecurringScheduler:
def __init__(self):
self.jobs = []
self.running = True
def every(self, name, seconds, func):
self.jobs.append({
'name': name,
'interval': seconds,
'func': func,
'last_run': 0,
})
def start(self):
def loop():
while self.running:
now = time.time()
for job in self.jobs:
if now - job['last_run'] >= job['interval']:
try:
job['func']()
except Exception as e:
print(f"Error in {job['name']}: {e}")
job['last_run'] = now
time.sleep(1)
threading.Thread(target=loop, daemon=True).start()
def stop(self):
self.running = False
def job_count(self):
return len(self.jobs)
sched = IntervalRecurringScheduler()
sched.every('health_check', 10, lambda: print(f" Health OK at {time.strftime('%H:%M:%S')}"))
sched.every('metrics', 30, lambda: print(f" Metrics collected"))
sched.start()
time.sleep(25)
sched.stop()
Expected output:
Health OK at 10:00:00
Health OK at 10:00:10
Health OK at 10:00:20
Metrics collected at 10:00:20
Database-Backed Dynamic Scheduler
import sqlite3
import time
import json
import threading
from datetime import datetime
class DatabaseScheduler:
def __init__(self, db_path=':memory:'):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self.conn.execute('''
CREATE TABLE IF NOT EXISTS schedules (
name TEXT PRIMARY KEY,
cron_expr TEXT,
task_type TEXT,
enabled INTEGER DEFAULT 1,
last_run TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
self.conn.commit()
self.running = True
def add_schedule(self, name, cron_expr, task_type):
self.conn.execute(
'INSERT OR REPLACE INTO schedules (name, cron_expr, task_type, enabled) VALUES (?, ?, ?, 1)',
(name, cron_expr, task_type)
)
self.conn.commit()
def disable_schedule(self, name):
self.conn.execute(
'UPDATE schedules SET enabled = 0 WHERE name = ?', (name,)
)
self.conn.commit()
def get_due_jobs(self):
cursor = self.conn.execute(
'SELECT name, cron_expr, task_type, last_run FROM schedules WHERE enabled = 1'
)
now = datetime.utcnow()
due = []
for name, cron, task_type, last_run in cursor:
schedule_key = f"{now.minute} {now.hour} {now.day} {now.month} {now.weekday()}"
if not last_run or self._is_due(cron, now, last_run):
due.append({'name': name, 'task_type': task_type})
return due
def _is_due(self, cron_expr, now, last_run_str):
parts = cron_expr.split()
if len(parts) != 5:
return False
if last_run_str:
last = datetime.fromisoformat(last_run_str)
if (now - last).total_seconds() < 55:
return False
return True
def mark_run(self, name):
self.conn.execute(
'UPDATE schedules SET last_run = ? WHERE name = ?',
(datetime.utcnow().isoformat(), name)
)
self.conn.commit()
def list_schedules(self):
cursor = self.conn.execute('SELECT name, cron_expr, enabled FROM schedules')
return cursor.fetchall()
dbs = DatabaseScheduler()
dbs.add_schedule('nightly_backup', '0 3 * * *', 'backup')
dbs.add_schedule('hourly_cleanup', '0 * * * *', 'cleanup')
print(dbs.list_schedules())
print(dbs.get_due_jobs())
Expected output:
[('nightly_backup', '0 3 * * *', 1), ('hourly_cleanup', '0 * * * *', 1)]
[]
Overlap Prevention
import time
import threading
class NonOverlappingScheduler:
def __init__(self):
self.running_jobs = set()
self._lock = threading.Lock()
def run_if_idle(self, name, func):
with self._lock:
if name in self.running_jobs:
print(f"Skipping {name}: previous instance still running")
return False
self.running_jobs.add(name)
try:
print(f"Starting {name}")
func()
return True
finally:
with self._lock:
self.running_jobs.discard(name)
print(f"Finished {name}")
def get_running(self):
with self._lock:
return list(self.running_jobs)
def slow_backup():
time.sleep(5)
print(" Backup done")
sched = NonOverlappingScheduler()
sched.run_if_idle('backup', slow_backup)
sched.run_if_idle('backup', slow_backup)
Expected output:
Starting backup
Skipping backup: previous instance still running
Backup done
Finished backup
Common Mistakes
1. No Overlap Protection
Recurring jobs that run longer than their interval pile up multiple instances. Use locking to prevent concurrent execution of the same job.
2. Hardcoded Schedules
Schedule changes require code deploys. Store schedules in the database or a config file to enable runtime changes without restarts.
3. Ignoring Timezone Changes
DST transitions cause jobs to run twice or not at all. Use UTC for scheduling and convert to local time only for display.
4. No Missed Schedule Handling
When the scheduler is down, missed jobs are lost. Implement catch-up logic that runs missed jobs on restart within a configurable window.
5. Single Point of Failure
One scheduler instance is a single point of failure. Use leader election or multiple schedulers with distributed locks for high availability.
Practice Questions
1. What is the difference between cron and interval scheduling?
Cron runs at specific calendar times (3 AM daily). Interval runs every N seconds regardless of calendar time.
2. How do you prevent overlapping recurring jobs?
Use a distributed lock (Redis lock, database advisory lock) that prevents a second instance from starting while the first is running.
3. Why use database-backed schedules?
Schedules become dynamic: add, remove, or modify them at runtime without code deploys. Enables self-service scheduling for users.
4. How do you handle DST transitions?
Use UTC for all scheduling logic. Convert to local time for display only. The scheduler never observes local time.
Challenge
Build a recurring scheduling system that supports cron and interval schedules, database-backed dynamic configuration, overlap prevention with Redis locks, catch-up logic for missed schedules within a 1-hour window, and metrics for schedule health.
FAQ
Mini Project: Recurring Scheduler System
import time
import threading
import sqlite3
from datetime import datetime
class RecurringScheduler:
def __init__(self):
self.conn = sqlite3.connect(':memory:')
self.conn.execute('''
CREATE TABLE schedules (
id INTEGER PRIMARY KEY,
name TEXT UNIQUE,
interval_seconds INTEGER,
task TEXT,
enabled INTEGER DEFAULT 1,
last_run REAL,
created_at TEXT
)
''')
self.conn.commit()
self.running = True
def add(self, name, interval_seconds, task):
self.conn.execute(
'INSERT OR REPLACE INTO schedules (name, interval_seconds, task, last_run, created_at) VALUES (?, ?, ?, ?, ?)',
(name, interval_seconds, task, 0, datetime.utcnow().isoformat())
)
self.conn.commit()
def run_loop(self):
while self.running:
now = time.time()
cursor = self.conn.execute(
'SELECT name, interval_seconds, task, last_run FROM schedules WHERE enabled = 1'
)
for name, interval, task, last_run in cursor:
if now - last_run >= interval:
print(f"[{datetime.utcnow().isoformat()}] Running {name}")
self.conn.execute(
'UPDATE schedules SET last_run = ? WHERE name = ?',
(now, name)
)
self.conn.commit()
time.sleep(1)
def stop(self):
self.running = False
def list_active(self):
cursor = self.conn.execute('SELECT name, interval_seconds FROM schedules WHERE enabled = 1')
return cursor.fetchall()
sched = RecurringScheduler()
sched.add('heartbeat', 10, 'send_heartbeat')
sched.add('cleanup', 30, 'cleanup_temp')
t = threading.Thread(target=sched.run_loop, daemon=True)
t.start()
time.sleep(5)
print(sched.list_active())
sched.stop()
Expected output:
[2026-06-28T...] Running heartbeat
[{'name': 'heartbeat', 'interval_seconds': 10}, {'name': 'cleanup', 'interval_seconds': 30}]
What's Next
Now that you understand recurring scheduling, explore job uniqueness for preventing duplicate schedules, then learn about job chaining for sequential task execution.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro