Periodic Tasks with Celery Beat
In this tutorial, you will learn about Periodic Tasks with Celery Beat. We cover key concepts, practical examples, and best practices to help you master this topic.
Schedule periodic tasks in Celery using Celery Beat, configure cron-like schedules, use database-backed schedules, and manage periodic task lifecycles.
What You Learn
You will learn how to set up Celery Beat for scheduled task execution, define periodic tasks with crontab and interval schedules, use the Django database scheduler, and handle missed tasks.
Why It Matters
Many applications need tasks to run on a schedule: nightly database cleanup, hourly report generation, daily email digests. Celery Beat is the built-in scheduler that makes this reliable without external cron dependencies.
Real-World Use
DodaTech uses Celery Beat for daily malware signature updates at 3 AM, hourly cache warming, and weekly analytics report generation. Beat ensures these run on time even through deployments and restarts.
Celery Beat Setup
flowchart LR
Beat[Celery Beat
Scheduler] -->|Tasks| B[Broker]
B -->|Tasks| W[Worker]
Beat -->|Schedule| S[Schedule Store
File/Database]
style Beat fill:#f90,color:#fff
from celery import Celery
from celery.schedules import crontab
app = Celery('beat_demo', broker='redis://localhost:6379/0')
app.conf.update(
beat_schedule={
'cleanup-every-hour': {
'task': 'beat_demo.cleanup_logs',
'schedule': crontab(minute=0), # Every hour at minute 0
'args': (24,),
},
'report-daily': {
'task': 'beat_demo.generate_report',
'schedule': crontab(hour=3, minute=0), # Daily at 3 AM
'kwargs': {'type': 'daily'},
},
'health-check': {
'task': 'beat_demo.health_check',
'schedule': 300.0, # Every 5 minutes (float = seconds)
},
},
)
@app.task
def cleanup_logs(hours):
return f"Cleaned logs older than {hours} hours"
@app.task
def generate_report(type='daily'):
return f"Generated {type} report"
@app.task
def health_check():
return "System healthy"
# Start Celery Beat
celery -A beat_demo beat --loglevel=info
# Start worker
celery -A beat_demo worker --loglevel=info
# Or run both in one process
celery -A beat_demo worker --beat --loglevel=info
Crontab Schedules
from celery.schedules import crontab
# Every minute
crontab()
# Every hour at minute 0
crontab(minute=0)
# Daily at midnight
crontab(hour=0, minute=0)
# Every Monday at 8 AM
crontab(hour=8, minute=0, day_of_week=1)
# Weekdays at 9 AM and 5 PM
crontab(hour='9,17', minute=0, day_of_week='1-5')
# First day of every month at midnight
crontab(hour=0, minute=0, day_of_month=1)
# Every 30 minutes during business hours (9-17 weekdays)
crontab(minute='*/30', hour='9-17', day_of_week='1-5')
# Every 15 minutes
crontab(minute='*/15')
# Every Sunday at 2 AM
crontab(hour=2, minute=0, day_of_week=0)
Interval Schedules
from celery.schedules import crontab
from datetime import timedelta
app.conf.update(
beat_schedule={
# Using seconds
'every-30-seconds': {
'task': 'tasks.heartbeat',
'schedule': 30.0,
},
# Using timedelta
'every-5-minutes': {
'task': 'tasks.cache_warm',
'schedule': timedelta(minutes=5),
},
# Using crontab
'nightly': {
'task': 'tasks.daily_report',
'schedule': crontab(hour=2, minute=0),
},
}
)
Solar Schedules
Run tasks based on sunrise/sunset times:
from celery.schedules import solar
app.conf.update(
beat_schedule={
'at-sunset': {
'task': 'tasks.night_mode',
'schedule': solar('sunset', 40.7128, -74.0060), # NYC
'kwargs': {'latitude': 40.7128, 'longitude': -74.0060},
},
'at-sunrise': {
'task': 'tasks.day_mode',
'schedule': solar('sunrise', 40.7128, -74.0060),
},
}
)
Solar event types: dawn_astronomical, dawn_civil, dawn_nautical, sunrise, solar_noon, sunset, dusk_civil, dusk_nautical, dusk_astronomical.
Database-Backed Scheduler
For dynamic schedules that change at runtime, use the database scheduler:
pip install celery[redis] django-celery-beat
# settings.py
INSTALLED_APPS = [
'django_celery_beat',
]
# celery.py
from celery import Celery
app = Celery('django_app')
app.config_from_object('django.conf:settings', namespace='CELERY')
# Use database scheduler
app.conf.beat_scheduler = 'django_celery_beat.schedulers:DatabaseScheduler'
# Django admin creates/modifies schedules at runtime
from django_celery_beat.models import PeriodicTask, CrontabSchedule
schedule, _ = CrontabSchedule.objects.get_or_create(
minute='0',
hour='3',
day_of_week='*',
day_of_month='*',
month_of_year='*',
)
PeriodicTask.objects.create(
crontab=schedule,
name='Cleanup-old-logs',
task='tasks.cleanup_logs',
args=[48],
)
Beat Entrypoint Script
# run_beat.py
from celery import Celery
from celery.schedules import crontab
import os
app = Celery('scheduler', broker=os.environ.get('BROKER_URL', 'redis://localhost:6379/0'))
app.conf.update(
beat_schedule={
'scan-directories': {
'task': 'tasks.scan_directories',
'schedule': crontab(minute='*/30'),
'options': {'queue': 'batch'},
},
'update-signatures': {
'task': 'tasks.update_virus_signatures',
'schedule': crontab(hour=2, minute=0),
'options': {'priority': 9},
},
'send-daily-digest': {
'task': 'tasks.send_daily_digest',
'schedule': crontab(hour=8, minute=0),
'kwargs': {'timezone': 'UTC'},
},
'healthcheck': {
'task': 'tasks.healthcheck',
'schedule': 60.0,
'options': {'expires': 55},
},
},
beat_max_loop_interval=5,
beat_scheduler='celery.beat:PersistentScheduler',
beat_schedule_filename='/var/run/celery/beat-schedule',
)
if __name__ == '__main__':
app.Beat().run()
Common Mistakes
1. Not Running Beat Separately
Beat must be running to schedule periodic tasks. A common mistake is starting only the worker and wondering why scheduled tasks never execute.
2. Using Timezone-Aware Schedules Without Setting timezone
If your crontab specifies hour=3 but timezone is UTC, it runs at 3 AM UTC, not 3 AM local time. Set timezone in Celery config.
3. Overlapping Task Executions
If a task takes longer than its schedule interval, multiple instances pile up. Use @app.task(lock=True) or set beat_max_loop_interval appropriately.
4. Not Handling Missed Tasks
If Beat is down for an hour, tasks scheduled during that window are skipped by default. Use beat_scheduler with persistent state to track missed runs.
5. Hardcoding Schedules in Code
For production, use database-backed scheduler so schedules can be changed without redeploying code.
Practice Questions
1. What is Celery Beat?
The built-in scheduler that sends periodic tasks to the Celery worker at configured intervals. It runs as a separate Process.
2. How do you schedule a task every 30 minutes?
Use crontab(minute='*/30') or schedule=1800.0 for interval-based scheduling.
3. What is the difference between PersistentScheduler and DatabaseScheduler?
PersistentScheduler stores schedules in a local file. DatabaseScheduler stores them in a database, allowing runtime modification via Django admin.
4. How do you prevent overlapping task executions?
Set a task lock (e.g., Redis lock) that prevents a second instance from starting if the first is still running. Celery does not prevent overlaps by default.
Challenge
Design a Celery Beat schedule for a security monitoring platform: vulnerability scans (daily at 2 AM), signature updates (hourly), threat feed ingestion (every 15 minutes), report generation (weekly on Monday 8 AM), and cleanup (every Sunday 3 AM). Include timezone handling and overlap prevention.
FAQ
Mini Project: Scheduled Task System
# scheduled_tasks.py
from celery import Celery
from celery.schedules import crontab
from datetime import timedelta
import os
app = Celery('scheduled_demo', broker='redis://localhost:6379/0')
app.conf.update(
timezone='UTC',
enable_utc=True,
beat_schedule={
'heartbeat': {
'task': 'scheduled_tasks.heartbeat',
'schedule': timedelta(seconds=30),
'args': ('system',),
},
'hourly-cleanup': {
'task': 'scheduled_tasks.cleanup',
'schedule': crontab(minute=0),
'kwargs': {'max_age_hours': 24},
},
'daily-summary': {
'task': 'scheduled_tasks.summary',
'schedule': crontab(hour=23, minute=59),
'args': ('daily',),
},
'weekly-report': {
'task': 'scheduled_tasks.report',
'schedule': crontab(hour=8, minute=0, day_of_week=1),
},
},
)
@app.task
def heartbeat(component):
print(f"[{component}] Heartbeat OK at {__import__('datetime').datetime.utcnow()}")
return "alive"
@app.task
def cleanup(max_age_hours=24):
print(f"Cleaning up records older than {max_age_hours} hours")
return f"Cleaned {max_age_hours}h+ records"
@app.task
def summary(period='daily'):
print(f"Generating {period} summary report")
return f"{period.capitalize()} summary generated"
@app.task
def report():
print("Generating weekly report...")
return "Weekly report ready"
if __name__ == '__main__':
print("Periodic Tasks with Celery Beat")
print("=" * 40)
print("Scheduled tasks:")
for name, config in app.conf.beat_schedule.items():
print(f" {name}: {config['schedule']}")
print()
print("Start Beat:")
print(" celery -A scheduled_tasks beat --loglevel=info")
print("Start Worker:")
print(" celery -A scheduled_tasks worker --loglevel=info")
Expected output:
Periodic Tasks with Celery Beat
========================================
Scheduled tasks:
heartbeat: 0:00:30
hourly-cleanup: crontab(minute=0)
daily-summary: crontab(hour=23, minute=59)
weekly-report: crontab(hour=8, minute=0, day_of_week=1)
Start Beat:
celery -A scheduled_tasks beat --loglevel=info
Start Worker:
celery -A scheduled_tasks worker --loglevel=info
What's Next
Now that you understand periodic tasks, explore task result backend for storing and retrieving task results, then learn about task chaining for building multi-step workflows.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro