Recurring Jobs and Periodic Tasks
In this tutorial, you will learn about Recurring Jobs and Periodic Tasks. We cover key concepts, practical examples, and best practices to help you master this topic.
Implement recurring jobs that run at fixed intervals or schedules using cron, interval timers, and calendar-based triggers for automated periodic work.
What You Learn
You will learn how to set up recurring jobs with interval and cron schedules, handle job persistence across restarts, manage timezone-aware scheduling, and avoid common periodic job pitfalls.
Why It Matters
Every backend system needs periodic tasks: database backups, cache warming, log rotation, report generation. Manual execution is error-prone and unsustainable. Recurring jobs automate these tasks so they run reliably without human intervention.
Real-World Use
DodaTech's security platform runs vulnerability scans every 6 hours, updates threat signatures every 30 minutes, cleans expired session records every night at 3 AM, and generates weekly audit reports every Monday at 8 AM. All through recurring jobs.
Interval-Based Recurring Jobs
import time
import threading
class IntervalWorker:
def __init__(self):
self.tasks = []
self.running = True
def add_recurring(self, name, interval_seconds, func):
self.tasks.append({
'name': name,
'interval': interval_seconds,
'func': func,
'last_run': 0,
})
def start(self):
def loop():
while self.running:
now = time.time()
for task in self.tasks:
if now - task['last_run'] >= task['interval']:
task['last_run'] = now
try:
task['func']()
except Exception as e:
print(f"Task {task['name']} failed: {e}")
time.sleep(1)
thread = threading.Thread(target=loop, daemon=True)
thread.start()
def stop(self):
self.running = False
worker = IntervalWorker()
def clean_temp_files():
print(f" Cleaning temp files at {time.strftime('%H:%M:%S')}")
def warm_cache():
print(f" Warming cache at {time.strftime('%H:%M:%S')}")
worker.add_recurring('cleanup', 10, clean_temp_files)
worker.add_recurring('cache_warm', 25, warm_cache)
worker.start()
time.sleep(30)
worker.stop()
Expected output:
Cleaning temp files at 10:00:00
Cleaning temp files at 10:00:10
Warming cache at 10:00:15
Cleaning temp files at 10:00:20
Cleaning temp files at 10:00:30
Cron-Based Recurring Jobs
import time
from datetime import datetime
import croniter
class CronRecurring:
def __init__(self):
self.jobs = {}
def add_cron(self, name, cron_expr, func):
cron = croniter.croniter(cron_expr, datetime.now())
next_run = cron.get_next(datetime)
self.jobs[name] = {
'cron_expr': cron_expr,
'cron': cron,
'func': func,
'next_run': next_run,
}
def tick(self):
now = datetime.now()
for name, job in list(self.jobs.items()):
if now >= job['next_run']:
try:
job['func']()
except Exception as e:
print(f"Cron job {name} failed: {e}")
job['cron'] = croniter.croniter(
job['cron_expr'], now
)
job['next_run'] = job['cron'].get_next(datetime)
print(f" {name} next run at {job['next_run']}")
def hourly_backup():
print(f" Hourly backup at {datetime.now()}")
def daily_report():
print(f" Daily report at {datetime.now()}")
scheduler = CronRecurring()
scheduler.add_cron('backup', '0 * * * *', hourly_backup)
scheduler.add_cron('report', '0 9 * * *', daily_report)
for _ in range(3):
scheduler.tick()
time.sleep(1)
Storing Recurring Jobs in Redis
import redis
import json
import time
from datetime import datetime
r = redis.Redis()
class PersistentRecurringScheduler:
def __init__(self):
self.schedule_key = 'recurring_schedules'
def add_job(self, name, cron_expr, task_data):
schedule = {
'name': name,
'cron': cron_expr,
'task': task_data,
'enabled': True,
'last_run': None,
}
r.hset(self.schedule_key, name, json.dumps(schedule))
print(f"Added recurring job: {name} ({cron_expr})")
def remove_job(self, name):
r.hdel(self.schedule_key, name)
def list_jobs(self):
jobs = r.hgetall(self.schedule_key)
for name, data in jobs.items():
yield name.decode(), json.loads(data)
def disable_job(self, name):
data = r.hget(self.schedule_key, name)
if data:
job = json.loads(data)
job['enabled'] = False
r.hset(self.schedule_key, name, json.dumps(job))
def enable_job(self, name):
data = r.hget(self.schedule_key, name)
if data:
job = json.loads(data)
job['enabled'] = True
r.hset(self.schedule_key, name, json.dumps(job))
scheduler = PersistentRecurringScheduler()
scheduler.add_job('cleanup_temp', '0 */2 * * *', {'task': 'cleanup'})
scheduler.add_job('generate_report', '0 9 * * 1', {'task': 'report'})
for name, job in scheduler.list_jobs():
status = 'enabled' if job['enabled'] else 'disabled'
print(f" {name}: {job['cron']} ({status})")
scheduler.disable_job('cleanup_temp')
Celery Beat Recurring Jobs
# celery_schedule.py
from celery import Celery
from celery.schedules import crontab
app = Celery('tasks', broker='redis://localhost:6379')
app.conf.beat_schedule = {
'cleanup-every-6-hours': {
'task': 'tasks.cleanup_expired',
'schedule': crontab(hour='*/6'),
'args': (7,),
},
'report-monday-9am': {
'task': 'tasks.generate_weekly_report',
'schedule': crontab(hour=9, minute=0, day_of_week=1),
'args': ('weekly',),
},
'health-check-5-min': {
'task': 'tasks.health_check',
'schedule': 300.0,
'args': ('https://api.dodatech.com',),
},
}
@app.task
def cleanup_expired(days):
print(f"Cleaning records older than {days} days")
@app.task
def generate_weekly_report(report_type):
print(f"Generating {report_type} report")
@app.task
def health_check(url):
print(f"Checking health of {url}")
Missed Execution Handling
import time
from datetime import datetime, timedelta
class MissedExecutionHandler:
def __init__(self, max_catchup_minutes=30):
self.max_catchup = timedelta(minutes=max_catchup_minutes)
def should_execute(self, scheduled_time, last_execution_time):
now = datetime.now()
if scheduled_time > now:
return False
if last_execution_time and scheduled_time <= last_execution_time:
return False
time_since_scheduled = now - scheduled_time
if time_since_scheduled > self.max_catchup:
print(f"Skipping missed execution from {scheduled_time}")
return False
return True
def catch_up(self, scheduled_times, func):
missed = 0
for sched_time in scheduled_times:
if self.should_execute(sched_time, None):
func()
missed += 1
return missed
handler = MissedExecutionHandler(max_catchup_minutes=60)
scheduled = [
datetime.now() - timedelta(minutes=10),
datetime.now() - timedelta(minutes=5),
datetime.now() - timedelta(minutes=2),
]
def send_alert():
print(" Alert sent")
caught_up = handler.catch_up(scheduled, send_alert)
print(f"Caught up {caught_up} missed executions")
Common Mistakes
1. No Persistence Across Restarts
Recurring jobs defined in memory are lost when the scheduler restarts. Store schedules in a database or Redis for persistence.
2. Ignoring Daylight Saving Time
Cron Jobs scheduled at 2:30 AM may run twice or not at all during DST transitions. Use UTC for all schedules and convert for display only.
3. Overlapping Executions
A recurring job that takes longer than its interval will overlap with the next execution. Use a lock to prevent concurrent runs of the same job.
4. Missing Health Monitoring
A recurring job that silently fails is worse than no job at all. Add monitoring, alerts, and logging to every periodic task.
5. Hardcoding Schedules
Schedule changes require code redeploys. Make schedules configurable via environment variables or a database-backed scheduler.
Practice Questions
1. What is the difference between interval and cron scheduling?
Interval scheduling runs every N seconds regardless of time. Cron scheduling runs at specific calendar times (e.g., every day at 3 AM).
2. How do you persist recurring jobs across restarts?
Store job definitions in Redis, a database, or a configuration file. The scheduler reloads all jobs on startup.
3. What causes overlapping job executions?
When a job takes longer than its schedule interval. The next execution starts before the previous one finishes.
4. How do you handle missed executions after downtime?
Set a catch-up tolerance window. Executions within that window run immediately. Older missed executions are skipped.
Challenge
Build a recurring job scheduler for monitoring system health: check disk space every 5 minutes, check service availability every minute, rotate logs daily at midnight, send weekly summary every Friday at 5 PM, and generate monthly billing report on the 1st at 2 AM. All schedules must be stored in Redis and survive restarts.
FAQ
Mini Project: Recurring Job Manager
import redis
import json
import time
import threading
from datetime import datetime
r = redis.Redis()
class RecurringJobManager:
def __init__(self):
self.running = True
self.registry_key = 'recurring:registry'
def register(self, name, schedule_type, interval, task, enabled=True):
job = {
'name': name,
'type': schedule_type,
'interval': interval,
'task': task,
'enabled': enabled,
'last_run': 0,
}
r.hset(self.registry_key, name, json.dumps(job))
return name
def unregister(self, name):
r.hdel(self.registry_key, name)
def toggle(self, name, enabled=None):
data = r.hget(self.registry_key, name)
if data:
job = json.loads(data)
job['enabled'] = enabled if enabled is not None else not job['enabled']
r.hset(self.registry_key, name, json.dumps(job))
def get_all(self):
jobs = r.hgetall(self.registry_key)
result = []
for name, data in jobs.items():
result.append(json.loads(data))
return result
def start_worker(self):
def loop():
while self.running:
now = time.time()
for job in self.get_all():
if not job['enabled']:
continue
if now - job['last_run'] >= job['interval']:
job['last_run'] = now
r.hset(self.registry_key, job['name'], json.dumps(job))
task_data = json.dumps({'task': job['task'], 'name': job['name']})
r.lpush('recurring_tasks', task_data)
print(f"Dispatched: {job['name']}")
time.sleep(1)
thread = threading.Thread(target=loop, daemon=True)
thread.start()
def stop(self):
self.running = False
manager = RecurringJobManager()
manager.register('health_check', 'interval', 10, 'check_health')
manager.register('cache_warm', 'interval', 30, 'warm_cache')
manager.register('log_rotate', 'interval', 60, 'rotate_logs', enabled=False)
manager.start_worker()
time.sleep(25)
manager.toggle('health_check', enabled=False)
time.sleep(10)
manager.stop()
print("Manager stopped")
Expected output:
Dispatched: health_check
Dispatched: health_check
Dispatched: cache_warm
Dispatched: health_check
Manager stopped
What's Next
Now that you understand recurring jobs, explore job deduplication for preventing duplicate task execution, then learn about job dependencies for chaining related tasks.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro