Introduction
By day 39 of building a Python backend, you've likely hit the wall where some operations just can't run synchronously. Email sends, file processing, third-party API calls—these block your request handlers and frustrate users. Celery for background jobs in Python Backend Development is the industry standard answer, and it's what teams at scale use to move expensive work off the critical path.
In this article, we'll walk through setting up Celery with a reliable broker, handling retries gracefully, designing idempotent tasks, and scheduling recurring work with Celery Beat. By the end, you'll have the patterns in place to run background jobs in both Django and FastAPI without losing sleep over lost work.
Understanding the Broker: Redis vs RabbitMQ
Celery is a task queue: it needs somewhere to store jobs before workers pick them up. The broker is that message layer. Redis and RabbitMQ are the two production choices, and they have different trade-offs.
Redis is simpler to operate. It's a single binary, sits in memory (with optional persistence), and works out of the box. Use Redis when your cluster is small, failure tolerance is moderate, and you want minimal ops overhead. It's what most startups and mid-size SaaS companies choose first.
RabbitMQ is a proper message broker. It offers durable queues, clustering, and acknowledgment semantics that guarantee delivery. Use RabbitMQ when you process critical financial transactions, need guaranteed delivery across network partitions, or run a large distributed system where a Redis crash would cascade.
Here's how to configure each in Celery:
# Redis broker (simple, fast)
from celery import Celery
app = Celery(
'myapp',
broker='redis://localhost:6379/0',
backend='redis://localhost:6379/1'
)
app.conf.update(
task_serializer='json',
accept_content=['json'],
result_serializer='json',
timezone='UTC',
enable_utc=True,
)
# RabbitMQ broker (durable, clustered)
from celery import Celery
app = Celery(
'myapp',
broker='amqp://guest:guest@localhost:5672//',
backend='rpc://' # or Redis for result backend
)
app.conf.update(
task_serializer='json',
accept_content=['json'],
result_serializer='json',
broker_connection_retry_on_startup=True,
broker_connection_retry=True,
broker_connection_max_retries=10,
)
Notice the backend setting. The broker queues tasks; the backend stores results. Many teams use Redis for both because it's convenient, but you can split them (RabbitMQ for durability, Redis for fast result lookups) depending on scale.
Retries with Exponential Backoff
Tasks fail. Networks drop, databases lock, external APIs timeout. Blindly retrying immediately leads to thundering herds and cascading failures. Exponential backoff spreads retries over time, giving the system breathing room to recover.
Here's a task that sends an invoice email with smart retry logic:
from celery import shared_task
from celery.exceptions import MaxRetriesExceededError
from django.core.mail import send_mail
import logging
logger = logging.getLogger(__name__)
@shared_task(bind=True, max_retries=5)
def send_invoice_email(self, invoice_id, recipient_email):
"""Send invoice PDF via email with exponential backoff retries."""
try:
invoice = Invoice.objects.get(id=invoice_id)
pdf_bytes = generate_invoice_pdf(invoice)
send_mail(
subject=f'Invoice {invoice.number}',
message='See attached PDF',
from_email='billing@company.com',
recipient_list=[recipient_email],
fail_silently=False,
)
logger.info(f'Invoice email sent for invoice {invoice_id}')
return {'status': 'sent', 'invoice_id': invoice_id}
except Exception as exc:
# Exponential backoff: 60s, 120s, 240s, 480s, 960s
countdown = 60 * (2 ** self.request.retries)
try:
raise self.retry(exc=exc, countdown=countdown)
except MaxRetriesExceededError:
logger.error(
f'Failed to send invoice {invoice_id} after 5 retries',
exc_info=exc
)
# Mark invoice as failed in your database
invoice = Invoice.objects.get(id=invoice_id)
invoice.email_status = 'failed'
invoice.save()
raise
The key is countdown: it tells Celery to wait before retrying. The formula 60 * (2 ** retries) doubles the wait each time. After 5 attempts (0, 1, 2, 3, 4), you've waited 16 minutes total. If the issue still isn't resolved, you log it and mark the record for manual review.
You can also use task routing to isolate failing tasks:
app.conf.task_routes = {
'myapp.tasks.send_invoice_email': {'queue': 'high_priority'},
'myapp.tasks.generate_report': {'queue': 'batch'},
}
app.conf.task_default_queue = 'default'
Run separate worker processes for each queue so a backlog of heavy reports doesn't starve time-sensitive emails.
Idempotent Tasks: Design for Duplicates
Here's an uncomfortable truth: distributed systems can execute the same task twice. Your result backend crashes after a task completes but before acknowledging the broker. The worker retries. You've now deducted the customer's balance twice.
The fix is idempotency: running a task twice produces the same result as running it once. This requires careful state design.
Bad approach:
@shared_task
def credit_account(user_id, amount):
"""DO NOT USE: Not idempotent."""
user = User.objects.get(id=user_id)
user.balance += amount
user.save()
return user.balance
If this runs twice, the balance increases twice. Disaster.
Good approach: use a unique task ID and check for duplicates:
import uuid
from django.db import models
class TaskRecord(models.Model):
"""Track executed tasks to prevent duplicates."""
task_id = models.CharField(max_length=255, unique=True, db_index=True)
task_name = models.CharField(max_length=255)
input_hash = models.CharField(max_length=64) # Hash of inputs
result = models.JSONField(null=True)
created_at = models.DateTimeField(auto_now_add=True)
@shared_task(bind=True)
def credit_account(self, user_id, amount):
"""Idempotent account credit."""
import hashlib
# Generate deterministic input hash
input_hash = hashlib.sha256(
f'{user_id}:{amount}'.encode()
).hexdigest()
# Check if already done
record, created = TaskRecord.objects.get_or_create(
task_id=self.request.id,
defaults={
'task_name': self.name,
'input_hash': input_hash,
}
)
if not created and record.result:
# Task already executed successfully
return record.result
# Execute the actual work
user = User.objects.get(id=user_id)
user.balance += amount
user.save()
result = {
'user_id': user_id,
'new_balance': user.balance,
'amount_credited': amount,
}
# Store result for idempotency check
record.result = result
record.save()
return result
Alternatively, use database constraints:
class CreditTransaction(models.Model):
task_id = models.CharField(max_length=255, unique=True)
user = models.ForeignKey(User, on_delete=models.CASCADE)
amount = models.DecimalField(max_digits=10, decimal_places=2)
created_at = models.DateTimeField(auto_now_add=True)
@shared_task(bind=True)
def credit_account(self, user_id, amount):
"""Idempotent via unique constraint."""
try:
# Create transaction with task_id as unique key
txn = CreditTransaction.objects.create(
task_id=self.request.id,
user_id=user_id,
amount=amount,
)
user = User.objects.get(id=user_id)
user.balance += amount
user.save()
return {'status': 'credited', 'transaction_id': txn.id}
except IntegrityError:
# Task already ran; fetch the result
txn = CreditTransaction.objects.get(task_id=self.request.id)
return {'status': 'already_credited', 'transaction_id': txn.id}
This approach leans on your database's ACID properties to guarantee idempotency. If the task runs twice, the second insert hits the unique constraint and you return the existing result.
Celery Beat for Scheduled Jobs
Celery Beat is Celery's scheduler. It runs in a separate process and enqueues tasks at fixed intervals or cron schedules. Common use cases: billing cycles, data cleanup, report generation, cache warming.
Configure Beat with periodic tasks:
from celery import Celery
from celery.schedules import crontab
app = Celery('myapp')
app.conf.beat_schedule = {
'generate-daily-report': {
'task': 'myapp.tasks.generate_daily_report',
'schedule': crontab(hour=0, minute=0), # Midnight UTC
'args': ('reports@company.com',),
},
'cleanup-old-sessions': {
'task': 'myapp.tasks.cleanup_old_sessions',
'schedule': 3600.0, # Every hour in seconds
},
'charge-monthly-subscriptions': {
'task': 'myapp.tasks.charge_subscriptions',
'schedule': crontab(day_of_month=1, hour=2, minute=0), # 1st of month, 2am
},
}
Start the Beat scheduler in production (usually on a single machine to avoid duplicate runs):
# shell command
celery -A myapp beat --loglevel=info
In development, use the in-process scheduler:
app.conf.beat_scheduler = 'django_celery_beat.schedulers:DatabaseScheduler'
This stores schedules in the database, so you can modify them without restarting.
A real scheduled task:
@shared_task
def charge_subscriptions():
"""Charge active subscriptions. Runs 1st of month at 2am UTC."""
subscriptions = Subscription.objects.filter(
status='active',
next_billing_date__lte=timezone.now()
)
failed_ids = []
for sub in subscriptions:
try:
charge_result = stripe.Charge.create(
amount=int(sub.amount_cents),
currency='usd',
customer=sub.stripe_customer_id,
)
sub.last_charged_at = timezone.now()
sub.next_billing_date = timezone.now() + timedelta(days=30)
sub.save()
except stripe.CardError as e:
failed_ids.append(sub.id)
logger.warning(f'Charge failed for subscription {sub.id}: {e}')
logger.info(f'Charged {len(subscriptions) - len(failed_ids)} subscriptions. Failed: {len(failed_ids)}')
return {
'total': subscriptions.count(),
'failed': len(failed_ids),
'failed_ids': failed_ids,
}
Common Mistakes
Not handling task idempotency. Tasks run twice; if yours isn't idempotent, you'll charge customers twice, send duplicate emails, or corrupt data. Always design for retries.
Using the same Redis instance for cache and Celery. A cache flush wipes your entire task queue. Separate them or use different databases within Redis.
Setting retries too high or backoff too short. Five retries with 60s backoff is reasonable. Thirty retries with 1s backoff hammers external APIs and fills logs. Be conservative.
Ignoring task timeouts. A task that hangs ties up a worker forever. Always set time_limit and soft_time_limit:
@shared_task(
bind=True,
max_retries=3,
time_limit=300, # 5 minutes hard limit
soft_time_limit=250, # 4 min 10 sec warning
)
def long_running_task(self):
# Handle soft timeout gracefully
pass
Not monitoring task queues. Tasks disappearing into the queue is a silent killer. Use Flower (Celery's web UI) or export metrics to Prometheus.
Key Takeaways
- Choose Redis for simplicity (most teams), RabbitMQ for durability and guaranteed delivery.
- Implement exponential backoff retries; don't hammer failing systems.
- Design all tasks as idempotent: duplicate execution must produce the same result.
- Use Celery Beat for scheduled work; store schedules in the database for easy updates.
- Monitor queues and worker health; silent task failures are worse than loud ones.
- Separate queues by priority so heavy batch jobs don't starve real-time tasks.
- Test task retries and failures in staging before production.
Comments
Loading comments…