Mini Project: Celery Order Processing Pipeline
In this tutorial, you will learn about Mini Project: Celery Order Processing Pipeline. We cover key concepts, practical examples, and best practices to help you master this topic.
Build a production-ready order processing pipeline with Celery using chained tasks, error handling, monitoring, and Django integration for a complete e-commerce backend.
What You Learn
You will build a complete order processing system: validate orders, Process payments, update inventory, send notifications, and generate receipts. This project combines chains, error handling, result backends, and monitoring.
Why It Matters
Order processing is a classic Celery use case. It involves multiple sequential steps, external API calls, error handling, and user notifications. Completing this project proves you can build production-quality Celery systems.
Real-World Use
This is the same pattern DodaTech uses for processing subscriptions. When a user purchases Durga Antivirus Pro, Celery validates the payment, provisions the license, updates inventory, and sends the welcome email, all asynchronously.
System Architecture
flowchart LR
P[Order Submitted] --> V[Validate Order]
V --> PA[Process Payment]
PA --> UI[Update Inventory]
UI --> GR[Generate Receipt]
GR --> SN[Send Notification]
SN --> D[Done]
PA -.-> |Failure| DL[Dead Letter Queue]
style PA fill:#f90,color:#fff
Project Structure
order_processing/
__init__.py
celery_app.py # Celery app instance
tasks.py # Task definitions
models.py # Data models
producer.py # Task submission
monitor.py # Monitoring script
requirements.txt # Dependencies
Step 1: Celery App Configuration
# celery_app.py
from celery import Celery
app = Celery('order_processing',
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,
task_track_started=True,
task_acks_late=True,
task_reject_on_worker_lost=True,
worker_prefetch_multiplier=1,
result_expires=86400,
task_routes={
'order_processing.tasks.*': {'queue': 'orders'},
'order_processing.tasks.send_*': {'queue': 'notifications'},
},
task_queues={
'orders': {'routing_key': 'orders.#'},
'notifications': {'routing_key': 'notifications.#'},
'default': {'routing_key': 'default.#'},
},
)
Step 2: Data Models
# models.py
from dataclasses import dataclass, field, asdict
from datetime import datetime
from typing import List, Optional
import json
@dataclass
class OrderItem:
product_id: str
name: str
quantity: int
price: float
@dataclass
class Order:
order_id: str
user_id: str
user_email: str
items: List[OrderItem]
total: float
status: str = 'pending'
created_at: str = ''
payment_id: Optional[str] = None
receipt_url: Optional[str] = None
def to_json(self):
return json.dumps(asdict(self))
@classmethod
def from_json(cls, data):
if isinstance(data, str):
data = json.loads(data)
items = [OrderItem(**i) for i in data['items']]
return cls(**{**data, 'items': items})
Step 3: Task Definitions
# tasks.py
from celery import chain, group, chord
from celery_app import app
from models import Order, OrderItem
import time
import random
import json
DEAD_LETTER_QUEUE = []
# Payment providers (simulated)
PAYMENT_PROVIDERS = {
'visa': lambda amt: random.random() > 0.1,
'mastercard': lambda amt: random.random() > 0.15,
'paypal': lambda amt: random.random() > 0.05,
}
@app.task(bind=True)
def validate_order(self, order_data):
"""Validate order data."""
order = Order.from_json(order_data)
print(f"Validating order {order.order_id}")
if not order.items:
raise ValueError(f"Order {order.order_id} has no items")
if order.total <= 0:
raise ValueError(f"Order {order.order_id} has invalid total")
order.status = 'validated'
print(f"Order {order.order_id} validated: {len(order.items)} items, ${order.total:.2f}")
return order.to_json()
@app.task(bind=True, max_retries=3)
def process_payment(self, order_data):
"""Process payment through external provider."""
order = Order.from_json(order_data)
print(f"Processing payment for order {order.order_id}: ${order.total:.2f}")
provider = 'paypal' if order.total > 100 else 'visa'
processor = PAYMENT_PROVIDERS.get(provider)
if not processor(order.total):
error_msg = f"Payment failed for order {order.order_id}"
if self.request.retries < self.max_retries:
delay = 10 * (2 ** self.request.retries)
print(f"{error_msg}, retrying in {delay}s (attempt {self.request.retries + 1})")
raise self.retry(countdown=delay)
else:
print(f"{error_msg}, sending to dead letter")
DEAD_LETTER_QUEUE.append({
'order_id': order.order_id,
'step': 'payment',
'error': error_msg,
'retries': self.request.retries,
})
raise
order.payment_id = f"PAY_{order.order_id}_{int(time.time())}"
order.status = 'paid'
print(f"Payment processed: {order.payment_id}")
return order.to_json()
@app.task(bind=True)
def update_inventory(self, order_data):
"""Update inventory for ordered items."""
order = Order.from_json(order_data)
print(f"Updating inventory for order {order.order_id}")
for item in order.items:
print(f" Reserved {item.quantity}x {item.name} (product {item.product_id})")
time.sleep(0.1)
order.status = 'inventory_updated'
return order.to_json()
@app.task(bind=True)
def generate_receipt(self, order_data):
"""Generate receipt document."""
order = Order.from_json(order_data)
print(f"Generating receipt for order {order.order_id}")
time.sleep(0.5)
receipt_id = f"RCP_{order.order_id}"
order.receipt_url = f"https://receipts.dodatech.com/{receipt_id}"
order.status = 'receipt_generated'
print(f"Receipt generated: {order.receipt_url}")
return order.to_json()
@app.task(bind=True)
def send_notification(self, order_data):
"""Send order confirmation to customer."""
order = Order.from_json(order_data)
print(f"Sending notification to {order.user_email}")
print(f" To: {order.user_email}")
print(f" Subject: Order {order.order_id} confirmed")
print(f" Receipt: {order.receipt_url}")
print(f" Items: {len(order.items)}")
print(f" Total: ${order.total:.2f}")
time.sleep(0.3)
order.status = 'completed'
print(f"Notification sent to {order.user_email}")
return order.to_json()
@app.task
def process_order(order_data):
"""Main order processing pipeline using chain."""
pipeline = chain(
validate_order.s(),
process_payment.s(),
update_inventory.s(),
generate_receipt.s(),
send_notification.s(),
)
return pipeline.delay(order_data)
Step 4: Task Submission
# producer.py
from models import Order, OrderItem
from tasks import process_order
import uuid
import time
def create_sample_order(order_id, should_fail=False):
"""Create a sample order."""
items = [
OrderItem(product_id='PROD-001', name='Durga Antivirus Pro', quantity=1, price=49.99),
OrderItem(product_id='PROD-002', name='Doda Browser Premium', quantity=2, price=29.99),
]
if should_fail:
items = []
return Order(
order_id=order_id,
user_id='USER-42',
user_email='customer@example.com',
items=items,
total=sum(item.price * item.quantity for item in items),
created_at=time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),
)
def submit_orders(num_orders=5):
"""Submit multiple orders for processing."""
task_ids = []
for i in range(num_orders):
order_id = f"ORD-{uuid.uuid4().hex[:8].upper()}"
order = create_sample_order(order_id, should_fail=(i == 3))
result = process_order(order.to_json())
print(f"Submitted order {order_id}: task {result.id[:8]}")
task_ids.append((order_id, result))
return task_ids
if __name__ == '__main__':
print("Starting Order Processing Pipeline")
print("=" * 40)
print("Make sure Celery workers are running:")
print(" celery -A tasks worker --queues=orders,notifications,default --loglevel=info")
print()
task_ids = submit_orders(5)
print("\nWaiting for results...")
for order_id, result in task_ids:
try:
output = result.get(timeout=60)
print(f"Order {order_id}: SUCCESS")
except Exception as e:
print(f"Order {order_id}: FAILED - {e}")
Expected output:
Starting Order Processing Pipeline
========================================
Make sure Celery workers are running:
celery -A tasks worker --queues=orders,notifications,default --loglevel=info
Submitted order ORD-A1B2C3D4: task 550e8400
Submitted order ORD-E5F6G7H8: task 6ba7b810
Submitted order ORD-I9J0K1L2: task 6ba7b811
Submitted order ORD-M3N4O5P6: task 6ba7b812
Submitted order ORD-Q7R8S9T0: task 6ba7b813
Waiting for results...
Order ORD-A1B2C3D4: SUCCESS
Order ORD-E5F6G7H8: SUCCESS
Order ORD-I9J0K1L2: SUCCESS
Order ORD-M3N4O5P6: FAILED - ValueError('Order ORD-M3N4O5P6 has no items')
Order ORD-Q7R8S9T0: SUCCESS
Step 5: Monitoring
# monitor.py
from celery_app import app
from tasks import DEAD_LETTER_QUEUE
import redis
import time
import os
def monitor_orders():
client = redis.Redis.from_url('redis://localhost:6379/0')
while True:
os.system('clear')
print(f"Order Processing Monitor - {time.ctime()}")
print("=" * 60)
# Queue depths
for queue_name in ['orders', 'notifications', 'default']:
depth = client.llen(queue_name)
status = 'OK' if depth < 10 else 'WARN' if depth < 50 else 'CRIT'
print(f" Queue '{queue_name}': {depth} tasks [{status}]")
# Worker status
try:
i = app.control.inspect()
workers = i.ping() or {}
print(f"\n Workers: {len(workers)}")
for w in workers:
active = i.active() or {}
tasks = active.get(w, [])
print(f" {w}: {len(tasks)} active tasks")
except Exception as e:
print(f"\n Workers: Error - {e}")
# Dead letter queue
if DEAD_LETTER_QUEUE:
print(f"\n Dead Letter Queue: {len(DEAD_LETTER_QUEUE)} items")
for entry in DEAD_LETTER_QUEUE[-3:]:
print(f" {entry['order_id']}: {entry['error']}")
print("\n Send new orders: python producer.py")
time.sleep(5)
if __name__ == '__main__':
monitor_orders()
Running the System
# Terminal 1: Start Celery worker
celery -A tasks worker --queues=orders,notifications,default \
--concurrency=4 --loglevel=info --events
# Terminal 2: Start monitoring
python monitor.py
# Terminal 3: Submit orders
python producer.py
Common Mistakes
1. Not Defining Queue Routes Correctly
If tasks go to wrong queues, workers do not pick them up. Verify routing with celery -A app inspect active_queues on each worker.
2. Not Handling Payment Failures
Payment failures need special handling: alert the user, log the error, and never auto-retry more than 3 times. Send failed payments to a dead letter queue for manual review.
3. Using One Queue for Everything
Separate order processing from notifications. Order tasks are time-sensitive and need dedicated workers. Notifications can tolerate delays.
4. Not Setting Task Time Limits
A payment gateway timeout should not hang a worker forever. Set time limits per task and configure soft/hard time limits in Celery config.
5. Forgetting to Start Beat for Scheduled Tasks
If you add periodic tasks (e.g., daily payment reconciliation), start Celery Beat. Running workers without Beat means scheduled tasks never execute.
Practice Questions
1. Why use a chain for order processing?
Order processing is sequential: validate before payment, payment before inventory. A chain ensures each step runs after the previous one succeeds.
2. How would you handle a payment timeout?
Set time_limit and soft_time_limit on the payment task. Use self.retry(countdown=30) to retry. After max retries, move to dead letter queue.
3. What happens if inventory update succeeds but receipt generation fails?
The chain stops at receipt generation. The order status remains at 'inventory_updated'. The payment is already processed. Implement a compensating Transaction or manual review process.
4. How do you scale this system?
Separate queues per task type. Add more workers for slow steps (payment, receipt generation). Use gevent workers for I/O-bound notification tasks.
Challenge
Extend the system: add order cancellation (compensating tasks that reverse payments and release inventory), implement a dead letter queue with automatic reprocessing after 1 hour, add a dashboard showing order throughput and failure rates, and implement idempotency keys to prevent duplicate payments.
FAQ
What's Next
You completed the Celery order processing pipeline. Review background jobs to compare Celery with other task queue solutions, then explore message queue patterns for broader messaging architecture knowledge.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro