Django Cron Jobs: Celery Beat, django-cron & Management Commands
Django does not include a built-in task scheduler. This guide covers every major approach to running recurring tasks in Django applications: management commands with system cron, Celery Beat for distributed task queues, the django-cron package, APScheduler integration, and django-extensions jobs. Each section includes production-ready code you can adapt.
Management Commands + Crontab
The simplest approach: write a Django management command and schedule it with the system cron job. No additional packages needed. This is the right choice when you have a single server and straightforward scheduling needs.
Creating a Management Command
# myapp/management/commands/cleanup_sessions.py
from django.core.management.base import BaseCommand
from django.utils import timezone
from datetime import timedelta
from myapp.models import Session
class Command(BaseCommand):
help = 'Delete expired sessions older than 7 days'
def add_arguments(self, parser):
parser.add_argument(
'--dry-run',
action='store_true',
help='Show what would be deleted without deleting',
)
def handle(self, *args, **options):
cutoff = timezone.now() - timedelta(days=7)
expired = Session.objects.filter(updated_at__lt=cutoff)
if options['dry_run']:
self.stdout.write(f'Would delete {expired.count()} sessions')
return
count, _ = expired.delete()
self.stdout.write(
self.style.SUCCESS(f'Deleted {count} expired sessions')
)Adding to Crontab
Use a cron expression generator to build the schedule:
# Edit crontab crontab -e # Run cleanup every day at 3 AM 0 3 * * * cd /home/deploy/myproject && /home/deploy/venv/bin/python manage.py cleanup_sessions >> /var/log/django-cron.log 2>&1 # Send weekly digest every Monday at 8 AM 0 8 * * 1 cd /home/deploy/myproject && /home/deploy/venv/bin/python manage.py send_weekly_digest >> /var/log/django-cron.log 2>&1
Tip
Always use the full path to your virtualenv Python binary and cd into your project directory. Cron runs with a minimal environment, so missing PATH or DJANGO_SETTINGS_MODULE is the number one cause of failed Django cron jobs.
Celery Beat Periodic Tasks
Celery Beat is the standard for production Django scheduling. It uses Redis or RabbitMQ as a message broker, supports cron and interval schedules, and separates scheduling from execution so you can scale workers independently.
pip install celery[redis] django-celery-beat
Celery Configuration
# myproject/celery.py
import os
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()# myproject/__init__.py
from .celery import app as celery_app
__all__ = ('celery_app',)Schedule with crontab() vs timedelta()
Celery Beat supports two schedule types. Use crontab() for cron expression-style schedules and timedelta() for fixed intervals:
# settings.py
from celery.schedules import crontab
from datetime import timedelta
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_BEAT_SCHEDULE = {
'cleanup-expired-sessions': {
'task': 'myapp.tasks.cleanup_sessions',
'schedule': crontab(hour=3, minute=0), # Daily at 3 AM
},
'sync-inventory': {
'task': 'myapp.tasks.sync_inventory',
'schedule': timedelta(hours=2), # Every 2 hours
},
'send-daily-digest': {
'task': 'myapp.tasks.send_digest',
'schedule': crontab(hour=8, minute=0, day_of_week='mon-fri'),
'kwargs': {'segment': 'premium'},
},
'generate-reports': {
'task': 'myapp.tasks.generate_reports',
'schedule': crontab(hour=1, minute=30, day_of_month=1), # Monthly
},
}Task Definitions
# myapp/tasks.py
from celery import shared_task
from django.utils import timezone
from datetime import timedelta
@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def cleanup_sessions(self):
from myapp.models import Session
cutoff = timezone.now() - timedelta(days=7)
count, _ = Session.objects.filter(updated_at__lt=cutoff).delete()
return f'Deleted {count} expired sessions'
@shared_task
def sync_inventory():
from myapp.services import InventoryService
result = InventoryService.sync_all()
return f'Synced {result["updated"]} products'
@shared_task(bind=True, max_retries=3)
def send_digest(self, segment='all'):
try:
from myapp.services import EmailService
sent = EmailService.send_digest(segment)
return f'Sent digest to {sent} users'
except Exception as exc:
raise self.retry(exc=exc)Running Celery Beat
# Start Celery worker celery -A myproject worker --loglevel=info # Start Celery Beat scheduler (separate process) celery -A myproject beat --loglevel=info # Or combined (development only) celery -A myproject worker --beat --loglevel=info
Database-Backed Schedules (django-celery-beat)
The django-celery-beat package stores schedules in your database, letting you create and modify them through the Django admin:
# settings.py
INSTALLED_APPS = [
...
'django_celery_beat',
]
# Run migrations
# python manage.py migrate django_celery_beat
# Start beat with database scheduler
# celery -A myproject beat --scheduler django_celery_beat.schedulers:DatabaseSchedulerWhy Celery Beat?
Celery Beat separates scheduling from execution. The Beat process enqueues tasks, and any number of workers process them. This gives you retries, concurrency control, rate limiting, priority queues, and horizontal scaling out of the box.
django-cron Package
django-cron provides a lightweight scheduling framework without the complexity of Celery. You define cron classes in your app, and a single management command runs them on schedule.
pip install django-cron
Defining Cron Jobs
# myapp/cron.py
from django_cron import CronJobBase, Schedule
class CleanupSessionsCronJob(CronJobBase):
RUN_AT_TIMES = ['03:00']
schedule = Schedule(run_at_times=RUN_AT_TIMES)
code = 'myapp.cleanup_sessions'
def do(self):
from myapp.models import Session
from django.utils import timezone
from datetime import timedelta
cutoff = timezone.now() - timedelta(days=7)
count, _ = Session.objects.filter(updated_at__lt=cutoff).delete()
print(f'Deleted {count} expired sessions')
class SyncInventoryCronJob(CronJobBase):
RUN_EVERY_MINS = 120 # Every 2 hours
schedule = Schedule(run_every_mins=RUN_EVERY_MINS)
code = 'myapp.sync_inventory'
def do(self):
from myapp.services import InventoryService
InventoryService.sync_all()Configuration and Running
# settings.py
INSTALLED_APPS = [..., 'django_cron']
CRON_CLASSES = [
'myapp.cron.CleanupSessionsCronJob',
'myapp.cron.SyncInventoryCronJob',
]
# Add to system crontab (run every 5 minutes)
# */5 * * * * cd /path/to/project && python manage.py runcrons >> /var/log/django-cron.log 2>&1Note
django-cron still requires a system cron entry to trigger runcrons. The package handles schedule checking and prevents overlapping runs, but the initial trigger comes from crontab.
APScheduler Integration
django-apscheduler integrates the popular APScheduler library with Django. Job state is stored in your database, and it supports cron, interval, and date-based triggers.
pip install django-apscheduler
# settings.py INSTALLED_APPS = [..., 'django_apscheduler'] APSCHEDULER_DATETIME_FORMAT = 'N j, Y, f:s a' APSCHEDULER_RUN_NOW_TIMEOUT = 25 # seconds
Registering Jobs
# myapp/jobs.py
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from django_apscheduler.jobstores import DjangoJobStore
scheduler = BackgroundScheduler()
scheduler.add_jobstore(DjangoJobStore(), 'default')
def cleanup_expired_sessions():
from myapp.models import Session
from django.utils import timezone
from datetime import timedelta
cutoff = timezone.now() - timedelta(days=7)
Session.objects.filter(updated_at__lt=cutoff).delete()
# Register with cron trigger
scheduler.add_job(
cleanup_expired_sessions,
trigger=CronTrigger(hour=3, minute=0),
id='cleanup_sessions',
max_instances=1,
replace_existing=True,
)
scheduler.start()Limitation
APScheduler runs inside the Django process. In multi-process deployments (gunicorn with multiple workers), each worker starts its own scheduler, causing duplicate execution. Use Celery Beat for multi-worker or multi-server environments.
django-extensions Jobs
django-extensions includes a simple jobs framework. You define job classes in a jobs directory, and run them with management commands that map to schedule frequencies.
# myapp/jobs/daily/cleanup.py
from django_extensions.management.jobs import DailyJob
class Job(DailyJob):
help = 'Delete expired sessions'
def execute(self):
from myapp.models import Session
from django.utils import timezone
from datetime import timedelta
cutoff = timezone.now() - timedelta(days=7)
Session.objects.filter(updated_at__lt=cutoff).delete()# Run daily jobs python manage.py runjobs daily # Run hourly jobs python manage.py runjobs hourly # Crontab entries 0 3 * * * cd /path/to/project && python manage.py runjobs daily 0 * * * * cd /path/to/project && python manage.py runjobs hourly
Monitoring with CronJobPro
All the approaches above can fail silently. A management command that raises an unhandled exception, or a Celery worker that crashes, won't alert anyone unless you set up monitoring. CronJobPro integrates with Django in two ways:
HTTP Trigger (Replace Cron)
Create a Django view for each task and let CronJobPro call it on schedule:
# myapp/views.py
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from functools import wraps
import os
def require_cron_token(f):
@wraps(f)
def wrapper(request, *args, **kwargs):
token = request.headers.get('Authorization', '')
expected = f"Bearer {os.environ.get('CRON_SECRET', '')}"
if token != expected:
return JsonResponse({'error': 'Unauthorized'}, status=401)
return f(request, *args, **kwargs)
return wrapper
@csrf_exempt
@require_POST
@require_cron_token
def cron_cleanup(request):
from myapp.models import Session
from django.utils import timezone
from datetime import timedelta
cutoff = timezone.now() - timedelta(days=7)
count, _ = Session.objects.filter(updated_at__lt=cutoff).delete()
return JsonResponse({'deleted': count})Heartbeat Monitoring
Add a ping at the end of any task to verify it completed successfully:
import urllib.request
def ping_cronjobpro(monitor_id):
try:
url = f'https://cronjobpro.com/api/ping/{monitor_id}'
urllib.request.urlopen(url, timeout=10)
except Exception:
pass # Don't fail the job if ping failsFrequently Asked Questions
What is the best way to schedule tasks in Django?
Celery Beat is the most popular option for production Django apps. For simpler needs, management commands with system cron work well. Choose based on whether you need retries, distributed workers, and database-backed schedule management.
Do I need Celery for Django cron jobs?
No. Management commands with system crontab, django-cron, or APScheduler all work without Celery. However, Celery is recommended for production apps needing retries, distributed execution, and reliable task processing.
How does Celery Beat work with Django?
Celery Beat reads your periodic task configuration and enqueues tasks at the right time via Redis or RabbitMQ. Workers pick up and execute tasks. You define schedules using crontab() or timedelta() in settings, or through django-celery-beat database models.
Can I use APScheduler with Django?
Yes. The django-apscheduler package stores job state in the database and supports cron, interval, and date triggers. Be aware it runs in-process, so multi-worker deployments can cause duplicate execution.
How do I monitor Django cron jobs?
Use Celery Flower for Celery task monitoring, or add heartbeat pings to management commands. CronJobPro can trigger your Django endpoints on schedule with automatic monitoring, retries, and failure alerts.
Related Articles
Schedule tasks in Python with APScheduler, Celery Beat, and the schedule library.
Cron Expression CheatsheetQuick-reference guide to cron expression syntax with common schedule examples.
Cron Job Not Running? 12 Common CausesDiagnose and fix the most common reasons cron jobs fail silently.
Cron Job Monitoring Best PracticesHeartbeat checks, log aggregation, alerting, and dashboards for cron jobs.
Schedule your Django endpoints externally
CronJobPro calls your Django views on schedule with automatic retries, monitoring, and alerts. No pip packages to install. Free for up to 5 jobs.