Celery with Django — Complete Guide
In this tutorial, you will learn about Celery with Django. We cover key concepts, practical examples, and best practices to help you master this topic.
Integrate Celery with Django for background task processing, configure the Django database scheduler, and handle tasks within the Django ORM lifecycle.
What You Learn
You will learn how to configure Celery with Django, define tasks that use the Django ORM, use the Django database scheduler for dynamic periodic tasks, and deploy Celery workers alongside Django.
Why It Matters
Django is synchronous by design. Long-running operations like sending emails, generating reports, or processing uploads block the HTTP response. Celery moves these to background workers, keeping Django responsive and scalable.
Real-World Use
Doda Browser's Django backend uses Celery for all background work: malware analysis after file upload, daily digest emails, cache warming, and log cleanup. The Django admin manages scheduled tasks via django-celery-beat.
Django-Celery Setup
# celery.py in your Django project
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()
# settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/1'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'UTC'
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_TIME_LIMIT = 30 * 60
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
# __init__.py (ensure Celery loads with Django)
from .celery import app as celery_app
__all__ = ('celery_app',)
Defining Django Tasks
# myapp/tasks.py
from celery import shared_task
from django.core.mail import send_mail
from django.utils import timezone
from .models import ScanJob, ScanResult
@shared_task
def process_scan(scan_job_id):
"""Process a scan job using Django ORM."""
try:
job = ScanJob.objects.get(id=scan_job_id)
except ScanJob.DoesNotExist:
return f"ScanJob {scan_job_id} not found"
job.status = 'processing'
job.started_at = timezone.now()
job.save()
# Simulate scan
result_data = {'clean': True, 'threats': []}
ScanResult.objects.create(
scan_job=job,
result=result_data,
completed_at=timezone.now()
)
job.status = 'completed'
job.completed_at = timezone.now()
job.save()
return f"Scan {scan_job_id} completed"
@shared_task
def send_welcome_email(user_email):
"""Send welcome email using Django's email system."""
send_mail(
'Welcome to DodaTech',
'Thank you for joining!',
'noreply@dodatech.com',
[user_email],
fail_silently=False,
)
return f"Email sent to {user_email}"
Calling Django Tasks
# myapp/views.py
from django.http import JsonResponse
from .tasks import process_scan, send_welcome_email
def upload_file(request):
"""Upload endpoint that triggers background processing."""
job = ScanJob.objects.create(
user=request.user,
file=request.FILES['file'],
status='pending'
)
task = process_scan.delay(job.id)
return JsonResponse({
'job_id': job.id,
'task_id': task.id,
'status': 'processing',
})
def check_status(request, job_id):
"""Check scan job status."""
from celery.result import AsyncResult
job = ScanJob.objects.get(id=job_id)
if job.task_id:
result = AsyncResult(job.task_id)
return JsonResponse({
'job_id': job_id,
'task_status': result.status,
'job_status': job.status,
})
return JsonResponse({'job_id': job_id, 'status': 'no_task'})
def register(request):
"""User registration with background email."""
user = create_user(request.POST)
send_welcome_email.delay(user.email)
return JsonResponse({'status': 'registered'})
Django Database-Backed Periodic Tasks
pip install django-celery-beat
# settings.py
INSTALLED_APPS = [
'django_celery_beat',
# ... other apps
]
python manage.py migrate django_celery_beat
python manage.py runserver # Admin interface at /admin/
# Create periodic tasks programmatically
from django_celery_beat.models import PeriodicTask, CrontabSchedule
# Create a crontab schedule (every day at 3 AM)
schedule, _ = CrontabSchedule.objects.get_or_create(
minute='0',
hour='3',
day_of_week='*',
day_of_month='*',
month_of_year='*',
)
# Create periodic task
PeriodicTask.objects.create(
crontab=schedule,
name='Daily cleanup of old scan results',
task='myapp.tasks.cleanup_old_scans',
args=[30], # Clean scans older than 30 days
kwargs={},
enabled=True,
)
Task Tracking in Database
# myapp/models.py
from django.db import models
from django.utils import timezone
class ScanJob(models.Model):
STATUS_CHOICES = [
('pending', 'Pending'),
('processing', 'Processing'),
('completed', 'Completed'),
('failed', 'Failed'),
]
user = models.ForeignKey('auth.User', on_delete=models.CASCADE)
file = models.FileField(upload_to='uploads/')
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending')
task_id = models.CharField(max_length=255, blank=True, null=True)
created_at = models.DateTimeField(auto_now_add=True)
started_at = models.DateTimeField(blank=True, null=True)
completed_at = models.DateTimeField(blank=True, null=True)
error_message = models.TextField(blank=True, default='')
class Meta:
indexes = [
models.Index(fields=['status']),
models.Index(fields=['task_id']),
]
class ScanResult(models.Model):
scan_job = models.OneToOneField(ScanJob, on_delete=models.CASCADE, related_name='result')
result = models.JSONField(default=dict)
completed_at = models.DateTimeField(default=timezone.now)
Using Django Signals with Celery
# myapp/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import ScanJob
from .tasks import process_scan
@receiver(post_save, sender=ScanJob)
def scan_job_created(sender, instance, created, **kwargs):
"""Auto-submit Celery task when ScanJob is created."""
if created and instance.status == 'pending':
task = process_scan.delay(instance.id)
instance.task_id = task.id
instance.save(update_fields=['task_id'])
Transaction Safety
Tasks that use the ORM should handle transaction state:
from celery import shared_task
from django.db import transaction
@shared_task
def process_after_commit(data_id):
"""Run task after the current transaction commits."""
transaction.on_commit(lambda: actual_process.delay(data_id))
@shared_task
def actual_process(data_id):
from .models import DataModel
# By the time this runs, the transaction has committed
obj = DataModel.objects.get(id=data_id)
obj.process()
return f"Processed {data_id}"
Common Mistakes
1. Accessing Expired ORM Objects
By the time a task runs, the database state may have changed. Always re-fetch objects inside the task instead of passing serialized model instances.
2. Not Handling DoesNotExist
Database records can be deleted between task submission and execution. Always wrap ORM queries in try/except and handle missing objects gracefully.
3. Creating Transactions in Tasks
Celery tasks should not use @transaction.atomic decorators. They run outside the request-response cycle. Use transaction.on_commit() for post-commit actions.
4. Passing Model Instances as Task Args
Celery serializes task arguments. Model instances are not JSON-serializable. Pass primary keys and re-fetch inside the task.
5. Not Running Migrations for django-celery-beat
The database scheduler requires its own database tables. Run manage.py migrate django_celery_beat before using the admin interface.
Practice Questions
1. How do you share_task decorator work in Django?
@shared_task creates a task that does not require importing the Celery app instance. It is the recommended way to define tasks in Django.
2. Why pass primary keys instead of model instances?
Model instances are not serializable. Celery serializes arguments to JSON. Pass the object's ID and re-fetch from the database inside the task.
3. What is the purpose of autodiscover_tasks()?
It automatically discovers task modules in each Django app by looking for tasks.py files. Workers register all tasks without manual imports.
4. How do you ensure a task runs after a database transaction commits?
Use transaction.on_commit() to submit the task only after the current transaction successfully commits. Prevents tasks from running on rolled-back data.
Challenge
Build a Django + Celery video processing system: users upload videos, a Celery task transcodes to multiple formats, generates thumbnails, and updates the database. Implement task tracking with model fields, progress updates via result backend, and admin integration for monitoring.
FAQ
Mini Project: Django Celery Integration
# 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()
# myapp/tasks.py
from celery import shared_task
from django.core.mail import send_mail
from django.db import transaction
from django.utils import timezone
import time
@shared_task(bind=True, max_retries=3)
def process_document(self, document_id):
"""Process an uploaded document."""
from .models import Document
try:
doc = Document.objects.select_for_update().get(id=document_id)
except Document.DoesNotExist:
return f"Document {document_id} not found"
doc.status = 'processing'
doc.save()
try:
time.sleep(2)
word_count = len(doc.content.split())
doc.word_count = word_count
doc.status = 'completed'
doc.processed_at = timezone.now()
doc.save()
send_mail(
'Document Processed',
f'Your document "{doc.title}" has been processed.',
'noreply@dodatech.com',
[doc.uploaded_by.email],
)
return f"Document {document_id} processed: {word_count} words"
except Exception as exc:
doc.status = 'failed'
doc.save()
raise self.retry(exc=exc, countdown=10)
# myapp/views.py
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from .models import Document
from .tasks import process_document
@csrf_exempt
def upload_document(request):
if request.method == 'POST':
doc = Document.objects.create(
title=request.POST['title'],
content=request.POST['content'],
uploaded_by=request.user,
)
task = process_document.delay(doc.id)
return JsonResponse({
'document_id': doc.id,
'task_id': task.id,
'status': 'queued',
})
def document_status(request, document_id):
from celery.result import AsyncResult
doc = Document.objects.get(id=document_id)
status_data = {
'document_id': document_id,
'title': doc.title,
'status': doc.status,
'word_count': doc.word_count,
}
if doc.task_id:
task_result = AsyncResult(doc.task_id)
status_data['task_state'] = task_result.state
status_data['task_result'] = str(task_result.result) if task_result.ready() else None
return JsonResponse(status_data)
What's Next
Now that you understand Celery with Django, explore Celery performance tuning for optimizing throughput, then build the mini project: order processing pipeline to apply everything you learned.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro