Background Job Scheduling — Complete Guide
In this tutorial, you will learn about Background Job Scheduling. We cover key concepts, practical examples, and best practices to help you master this topic.
Schedule Background Jobs with cron expressions, interval-based timing, and calendar-based triggers for recurring maintenance and periodic tasks.
What You Learn
You will learn how to schedule jobs using cron expressions, interval timing, and calendar schedules, implement job scheduling in different queue systems, and handle missed schedules.
Why It Matters
Many tasks must run on a schedule: nightly database backups, hourly cache warming, weekly report generation. Proper scheduling ensures these tasks run reliably at the right time without manual intervention.
Real-World Use
DodaTech schedules malware signature updates every 2 hours, database backups at 3 AM, weekly analytics reports on Monday morning, and log rotation every 6 hours. The scheduling system handles timezone changes and daylight saving.
Cron Expressions
# Cron expression format:
# minute hour day-of-month month day-of-week
# Every minute
"* * * * *"
# Every hour at minute 0
"0 * * * *"
# Daily at midnight
"0 0 * * *"
# Every Monday at 9 AM
"0 9 * * 1"
# Every 15 minutes during business hours (9-17) on weekdays
"*/15 9-17 * * 1-5"
# First day of month at 2 AM
"0 2 1 * *"
# Every Sunday at midnight
"0 0 * * 0"
# Every 30 minutes
"*/30 * * * *"
# Twice a day (8 AM and 8 PM)
"0 8,20 * * *"
Cron Implementation
import time
from datetime import datetime
import croniter
class CronScheduler:
def __init__(self):
self.jobs = []
def add_job(self, cron_expr, func, name=None):
self.jobs.append({
'name': name or func.__name__,
'cron': croniter.croniter(cron_expr, datetime.now()),
'func': func,
'last_run': None,
})
def check(self):
now = datetime.now()
for job in self.jobs:
next_run = job['cron'].get_next(datetime)
if now >= next_run:
if job['last_run'] != next_run:
print(f"Running scheduled job: {job['name']}")
job['func']()
job['last_run'] = next_run
job['cron'] = croniter.croniter(
job['cron'].cron_string,
now
)
def backup_job():
print(" Backup completed")
def cleanup_job():
print(" Cleanup completed")
scheduler = CronScheduler()
scheduler.add_job("*/1 * * * *", backup_job, "hourly backup")
scheduler.add_job("*/2 * * * *", cleanup_job, "cleanup")
for _ in range(5):
scheduler.check()
time.sleep(10)
Interval Scheduling
import time
import threading
class IntervalScheduler:
def __init__(self):
self.jobs = []
self.running = True
def every(self, seconds):
def decorator(func):
self.jobs.append({'interval': seconds, 'func': func, 'last_run': 0})
return func
return decorator
def start(self):
def loop():
while self.running:
now = time.time()
for job in self.jobs:
if now - job['last_run'] >= job['interval']:
job['func']()
job['last_run'] = now
time.sleep(1)
threading.Thread(target=loop, daemon=True).start()
def stop(self):
self.running = False
scheduler = IntervalScheduler()
@scheduler.every(10)
def check_health():
print(f" Health check at {time.strftime('%H:%M:%S')}")
@scheduler.every(30)
def generate_report():
print(f" Report generated at {time.strftime('%H:%M:%S')}")
scheduler.start()
time.sleep(35)
scheduler.stop()
Expected output:
Health check at 10:00:00
Health check at 10:00:10
Health check at 10:00:20
Report generated at 10:00:20
Health check at 10:00:30
Calendar-Based Scheduling
from datetime import datetime, timedelta
import calendar
class CalendarScheduler:
def schedule_business_hours(self, task):
"""Schedule only during business hours (9-17 weekdays)."""
now = datetime.now()
if now.weekday() < 5 and 9 <= now.hour < 17:
task()
else:
print(f"Outside business hours, skipping: {task.__name__}")
def schedule_end_of_month(self, task):
"""Schedule for the last day of each month."""
now = datetime.now()
last_day = calendar.monthrange(now.year, now.month)[1]
if now.day == last_day:
task()
def schedule_first_monday(self, task):
"""Schedule for the first Monday of each month."""
now = datetime.now()
month_cal = calendar.monthcalendar(now.year, now.month)
first_monday = month_cal[0][0] if month_cal[0][0] > 0 else month_cal[1][0]
if now.day == first_monday:
task()
cal = CalendarScheduler()
def monthly_report():
print("Monthly report generated")
def payroll():
print("Payroll processed")
cal.schedule_end_of_month(payroll)
cal.schedule_first_monday(monthly_report)
Integration with Job Queues
import redis
import json
import time
from datetime import datetime
import threading
r = redis.Redis()
class ScheduledQueue:
def __init__(self):
self.running = True
def schedule(self, queue, job, cron_expr, job_id=None):
"""Schedule a job with cron expression."""
schedule_entry = {
'queue': queue,
'job': job,
'cron': cron_expr,
'job_id': job_id,
'next_run': time.time(),
}
r.set(f"schedule:{job_id or job.get('task')}", json.dumps(schedule_entry))
print(f"Scheduled: {job.get('task')} ({cron_expr})")
def process_schedules(self):
"""Check and execute scheduled jobs."""
while self.running:
keys = r.keys("schedule:*")
for key in keys:
entry = json.loads(r.get(key))
if time.time() >= entry['next_run']:
r.lpush(entry['queue'], json.dumps(entry['job']))
next_time = int(time.time()) + 60
entry['next_run'] = next_time
r.set(key, json.dumps(entry))
print(f"Enqueued scheduled: {entry['job'].get('task')}")
time.sleep(5)
def stop(self):
self.running = False
sq = ScheduledQueue()
sq.schedule('default', {'task': 'hourly_cleanup'}, '0 * * * *', 'cleanup-1')
sq.schedule('default', {'task': 'daily_backup'}, '0 3 * * *', 'backup-1')
t = threading.Thread(target=sq.process_schedules, daemon=True)
t.start()
time.sleep(10)
sq.stop()
Missed Schedule Handling
import time
from datetime import datetime, timedelta
class ReliableScheduler:
def __init__(self, tolerance_minutes=5):
self.tolerance = timedelta(minutes=tolerance_minutes)
def should_run(self, scheduled_time, last_run):
now = datetime.now()
if scheduled_time > now:
return False
if last_run and scheduled_time <= last_run:
return False
if now - scheduled_time > self.tolerance:
print(f"Skipping missed schedule from {scheduled_time}")
return False
return True
def catch_up(self, scheduled_time, func):
"""Run missed jobs that are within tolerance."""
if self.should_run(scheduled_time, None):
print(f"Catching up: {func.__name__} from {scheduled_time}")
func()
scheduler = ReliableScheduler(tolerance_minutes=30)
# Simulate system being down for 15 minutes
scheduled = datetime.now() - timedelta(minutes=15)
def backup():
print(" Backup executed")
scheduler.catch_up(scheduled, backup)
Common Mistakes
1. Using Client-Side Scheduling Only
Client-side timers stop when the application restarts. Use server-side scheduling (Celery Beat, cron, systemd timers) for production.
2. Not Handling Timezone Changes
Cron expressions in UTC behave differently in different timezones. Use UTC everywhere or handle DST transitions explicitly.
3. Overlapping Job Executions
If a job takes longer than its schedule interval, multiple instances pile up. Implement locking or skip if previous instance is still running.
4. Ignoring Missed Schedules
When the scheduler is down for maintenance, tasks scheduled during that window are skipped. Implement catch-up logic if needed.
5. Hardcoding Schedule Intervals
Schedule changes require code deploys. Use database-backed schedulers (django-celery-beat) that allow runtime changes.
Practice Questions
1. What are the three main scheduling approaches?
Cron expressions (fixed schedule), interval timing (every N seconds/minutes), and calendar-based (business hours, end of month).
2. How do you prevent overlapping job executions?
Use a lock (Redis lock, database lock) that prevents a second instance from starting if the first is still running.
3. What is the cron expression for every hour?
0 * * * * (run at minute 0 of every hour). Use */1 * * * * for every minute.
4. How do you handle missed schedules?
Track last run time. If the gap between scheduled time and current time exceeds a tolerance, either skip or catch up.
Challenge
Design a scheduling system for a security platform: vulnerability scans (daily at 2 AM, tolerance 1 hour), threat feed updates (every 30 minutes, no tolerance), report generation (first Monday of month at 8 AM), cleanup (every Sunday at 3 AM, skip if missed more than 24h).
FAQ
Mini Project: Job Scheduler
import time
import threading
import redis
import json
from datetime import datetime
r = redis.Redis()
class JobScheduler:
def __init__(self):
self.jobs = {}
self.running = True
def add_cron(self, name, cron_expr, task, data=None):
self.jobs[name] = {
'type': 'cron',
'schedule': self._parse_cron(cron_expr),
'task': task,
'data': data or {},
'last_run': 0,
}
def add_interval(self, name, seconds, task, data=None):
self.jobs[name] = {
'type': 'interval',
'interval': seconds,
'task': task,
'data': data or {},
'last_run': 0,
}
def _parse_cron(self, expr):
parts = expr.split()
return {
'minute': parts[0],
'hour': parts[1],
'day': parts[2],
'month': parts[3],
'weekday': parts[4],
}
def _matches_cron(self, cron, now):
def match(val, current):
return val == '*' or str(current) in val.split(',')
return (match(cron['minute'], now.minute) and
match(cron['hour'], now.hour) and
match(cron['day'], now.day) and
match(cron['month'], now.month) and
match(cron['weekday'], now.weekday()))
def run(self):
print("Scheduler started")
while self.running:
now = datetime.now()
for name, job in self.jobs.items():
if job['type'] == 'interval':
if time.time() - job['last_run'] >= job['interval']:
job['last_run'] = time.time()
r.lpush('scheduled', json.dumps({'task': job['task'], 'data': job['data']}))
print(f"Scheduled: {job['task']}")
elif job['type'] == 'cron':
if self._matches_cron(job['schedule'], now) and time.time() - job['last_run'] >= 55:
job['last_run'] = time.time()
r.lpush('scheduled', json.dumps({'task': job['task'], 'data': job['data']}))
print(f"Scheduled: {job['task']}")
time.sleep(5)
def stop(self):
self.running = False
scheduler = JobScheduler()
scheduler.add_interval('health', 15, 'health_check')
scheduler.add_cron('backup', '*/1 * * * *', 'backup_task')
t = threading.Thread(target=scheduler.run, daemon=True)
t.start()
time.sleep(20)
scheduler.stop()
Expected output:
Scheduler started
Scheduled: health_check
Scheduled: backup_task
Scheduled: health_check
What's Next
Now that you understand job scheduling, explore job priorities for controlling execution order, then learn about job retries and backoff for handling failures.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro