Celery: Distributed Task Queue for Python — Complete Guide

pip install celery[redis] flower

# tasks.py
# Celery app configuration
from celery import Celery

app = Celery(
    'tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/0',
    include=['tasks'],
)

app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='UTC',
    enable_utc=True,
)

# tasks.py (continued)
# A simple Celery task

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def send_welcome_email(self, user_email, username):
    """Send a welcome email to a new user."""
    try:
        # Simulate email sending
        print(f"Sending welcome email to {user_email}...")
        # import requests
        # response = requests.post(
        #     "https://api.sendgrid.com/v3/mail/send",
        #     json={"to": user_email, "template": "welcome"},
        #     headers={"Authorization": "Bearer SG_API_KEY"}
        # )
        # response.raise_for_status()
        print(f"Welcome email sent to {username} at {user_email}")
        return {"status": "sent", "email": user_email}
    except Exception as exc:
        print(f"Failed to send email: {exc}")
        raise self.retry(exc=exc)

@app.task
def process_image(image_path):
    """Generate thumbnails for an uploaded image."""
    import time
    time.sleep(2)  # Simulate processing
    print(f"Processed image: {image_path}")
    return {"image": image_path, "thumbnail": image_path.replace('.jpg', '_thumb.jpg')}

celery -A tasks worker --loglevel=info

# call_tasks.py
# Demonstrates calling Celery tasks synchronously and asynchronously
from tasks import send_welcome_email, process_image

# Async call — returns immediately
result = send_welcome_email.delay('user@example.com', 'Alice')
print(f"Task ID: {result.id}")
print(f"Task status: {result.status}")

# Wait for result (blocking)
print(f"Result: {result.get(timeout=10)}")

# Async call with arguments
image_result = process_image.delay('/uploads/photo.jpg')
print(f"Image task ID: {image_result.id}")

# Check status without blocking
from celery.result import AsyncResult
status = AsyncResult(image_result.id, app=send_welcome_email.app)
print(f"Pending: {status.status}")  # PENDING, STARTED, SUCCESS, FAILURE

# chaining.py
# Chain tasks so one runs after another with the previous result as input
from celery import chain, group, chord
from tasks import app

@app.task
def add(x, y):
    result = x + y
    print(f"{x} + {y} = {result}")
    return result

@app.task
def multiply(x, y):
    result = x * y
    print(f"{x} * {y} = {result}")
    return result

@app.task
def format_result(value):
    formatted = f"Final result: {value}"
    print(formatted)
    return formatted

# Chain: add(2, 2) → multiply(4, 10) → format_result("Final result: 40")
task_chain = chain(
    add.s(2, 2),
    multiply.s(10),
    format_result.s()
)
result = task_chain()
print(f"Chain result: {result.get()}")

# Group: run tasks in parallel
parallel = group(
    add.s(1, 2),
    add.s(3, 4),
    add.s(5, 6),
)
group_result = parallel()
print(f"Group results: {group_result.get()}")

# celerybeat_schedule.py
# Add to your Celery app configuration for periodic tasks
from celery.schedules import crontab

app.conf.beat_schedule = {
    'cleanup-expired-sessions': {
        'task': 'tasks.cleanup_sessions',
        'schedule': crontab(hour=3, minute=0),  # Daily at 3 AM
    },
    'generate-daily-report': {
        'task': 'tasks.generate_report',
        'schedule': crontab(hour=8, minute=30, day_of_week='1-5'),  # Weekdays 8:30
    },
    'health-check-every-5-min': {
        'task': 'tasks.health_check',
        'schedule': 300.0,  # Every 5 minutes (in seconds)
    },
}

@app.task
def cleanup_sessions():
    print("Cleaning up expired sessions...")
    # Database cleanup logic
    return "Cleaned 150 expired sessions"

# Start beat: celery -A tasks beat --loglevel=info

# Start Flower dashboard
celery -A tasks flower --port=5555 --broker=redis://localhost:6379/0

# error_handling.py
# Robust task with retry logic and dead-letter handling

@app.task(bind=True, max_retries=5, default_retry_delay=30, acks_late=True)
def process_payment(self, order_id, amount):
    """Process payment with exponential backoff retry."""
    try:
        # Simulate flaky payment gateway
        import random
        if random.random() < 0.3:  # 30% failure rate
            raise ConnectionError("Payment gateway timeout")

        print(f"Payment {order_id}: ${amount} processed successfully")
        return {"order_id": order_id, "status": "completed", "amount": amount}

    except ConnectionError as exc:
        # Exponential backoff: 30s, 60s, 120s, 240s, 480s
        countdown = 30 * (2 ** self.request.retries)
        print(f"Retry {self.request.retries + 1}/{self.max_retries} "
              f"for order {order_id} in {countdown}s")
        raise self.retry(exc=exc, countdown=countdown)

    except Exception as exc:
        # Non-retriable error — log and acknowledge
        print(f"Fatal error processing payment {order_id}: {exc}")
        # In production: send to dead-letter queue or alert
        raise

# Task expiry and soft time limits
@app.task(time_limit=30, soft_time_limit=25, acks_late=True)
def process_large_file(file_path):
    """Process a large file with timeout protection."""
    import time
    start = time.time()
    # Simulate processing
    time.sleep(10)
    elapsed = time.time() - start
    print(f"Processed {file_path} in {elapsed:.2f}s")
    return f"Completed: {file_path}"

# monitor.py
# Simple Celery task monitor using Redis
import time
from datetime import datetime
from tasks import app

def monitor_queues():
    """Monitor Celery queue lengths in real time."""
    import redis
    r = redis.Redis(host='localhost', port=6379, db=0)

    print("═" * 50)
    print("Celery Queue Monitor — Press Ctrl+C to exit")
    print("═" * 50)

    try:
        while True:
            # Get queue lengths
            celery_queue = r.llen('celery')
            scheduled = len(r.zrange('celery', 0, -1))

            # Get worker stats
            inspector = app.control.inspect()
            active = inspector.active() or {}
            reserved = inspector.reserved() or {}
            worker_count = len(active)

            total_active = sum(len(tasks) for tasks in active.values())
            total_reserved = sum(len(tasks) for tasks in reserved.values())

            print(f"\n[{datetime.now().strftime('%H:%M:%S')}] "
                  f"Workers: {worker_count} | "
                  f"Queued: {celery_queue} | "
                  f"Active: {total_active} | "
                  f"Reserved: {total_reserved} | "
                  f"Scheduled: {scheduled}")

            time.sleep(5)
    except KeyboardInterrupt:
        print("\nMonitoring stopped.")

if __name__ == '__main__':
    monitor_queues()