Celery in Production: Picking the Right Jobs, Writing Safe Tasks, Running It Like You Mean It¶

# Celery use cases
celery_use_cases = {
    "http_offloading": "Send emails, webhooks, notifications after HTTP response",
    "cpu_intensive": "Image processing, data transformation, ML inference",
    "scheduled_jobs": "Daily reports, cleanup tasks, periodic maintenance",
    "etl_microsteps": "Data validation, transformation, loading steps",
    "idempotent_side_effects": "File generation, cache warming, external API calls"
}

# When NOT to use Celery
not_celery_cases = {
    "in_request_io": "Use threads/asyncio for parallel I/O within request",
    "streaming_data": "Use Kafka/Beam for event-time windows and exactly-once",
    "olap_queries": "Use specialized OLAP engines for analytical workloads",
    "real_time": "Use WebSockets or streaming for real-time communication"
}

# app/__init__.py
from celery import Celery
import os

def make_celery() -> Celery:
    app = Celery(
        "myapp",
        broker=os.getenv("CELERY_BROKER_URL", "redis://localhost:6379/0"),
        backend=os.getenv("CELERY_RESULT_BACKEND", "redis://localhost:6379/1"),
        include=["app.tasks"],  # modules to autodiscover
    )
    app.conf.update(
        task_serializer="json",
        accept_content=["json"],
        result_serializer="json",
        timezone="UTC",
        task_acks_late=True,            # ack after task completes
        task_reject_on_worker_lost=True,# don't ghost-ack on crashes
        worker_max_tasks_per_child=1000,# avoid leaks
        broker_heartbeat=30,
    )
    return app

celery = make_celery()

# app/tasks.py
from . import celery
from celery import shared_task
import time, httpx

@shared_task(
    bind=True,
    autoretry_for=(httpx.HTTPError,),
    retry_backoff=2,            # 2, 4, 8, …
    retry_jitter=True,
    retry_kwargs={"max_retries": 5},
    time_limit=60,              # hard kill
    soft_time_limit=55,         # graceful timeout
    rate_limit="10/m",          # protect downstream
    acks_late=True,
)
def send_webhook(self, url: str, payload: dict, idem_key: str | None = None) -> dict:
    """
    Idempotent side-effect: POST with Idempotency-Key so retries don't duplicate work.
    """
    headers = {"Idempotency-Key": idem_key} if idem_key else {}
    with httpx.Client(timeout=10) as c:
        r = c.post(url, json=payload, headers=headers)
        r.raise_for_status()
        return {"status": r.status_code}

@shared_task
def ping() -> str:
    return "pong"

# Idempotent task with external key
@shared_task(bind=True, acks_late=True)
def process_payment(self, user_id: str, amount: int, idem_key: str) -> dict:
    """
    Idempotent payment processing with external key.
    """
    # Check if already processed
    existing = get_payment_by_idem_key(idem_key)
    if existing:
        return {"status": "already_processed", "payment_id": existing.id}

    # Process payment
    payment = create_payment(user_id, amount, idem_key)
    return {"status": "processed", "payment_id": payment.id}

# Bounded task with limits
@shared_task(
    bind=True,
    time_limit=300,           # 5 minutes hard limit
    soft_time_limit=270,     # 4.5 minutes graceful
    rate_limit="5/m",         # 5 per minute
    autoretry_for=(Exception,),
    retry_kwargs={"max_retries": 3}
)
def generate_report(self, report_type: str, user_id: str) -> dict:
    """
    Generate report with bounded execution.
    """
    start_time = time.time()

    try:
        # Generate report
        report_data = build_report(report_type, user_id)

        # Save to storage
        report_url = save_report_to_s3(report_data)

        return {
            "status": "completed",
            "report_url": report_url,
            "duration": time.time() - start_time
        }
    except Exception as exc:
        # Log error and retry
        self.retry(countdown=60, exc=exc)

# Observable task with structured logging
import logging
import json

logger = logging.getLogger(__name__)

@shared_task(bind=True, acks_late=True)
def process_data(self, data_id: str, operation: str) -> dict:
    """
    Observable data processing with structured logging.
    """
    task_id = self.request.id
    logger.info(json.dumps({
        "event": "task_started",
        "task_id": task_id,
        "data_id": data_id,
        "operation": operation,
        "timestamp": time.time()
    }))

    try:
        # Process data
        result = perform_operation(data_id, operation)

        logger.info(json.dumps({
            "event": "task_completed",
            "task_id": task_id,
            "data_id": data_id,
            "operation": operation,
            "result": result,
            "timestamp": time.time()
        }))

        return result

    except Exception as exc:
        logger.error(json.dumps({
            "event": "task_failed",
            "task_id": task_id,
            "data_id": data_id,
            "operation": operation,
            "error": str(exc),
            "timestamp": time.time()
        }))
        raise

# app/celery_routes.py
ROUTES = {
    "app.tasks.send_webhook": {"queue": "io"},
    "app.tasks.generate_report": {"queue": "cpu"},
    "app.tasks.send_email": {"queue": "io"},
    "app.tasks.process_image": {"queue": "cpu"},
}

# app/__init__.py (add to make_celery)
app.conf.task_routes = "app.celery_routes.ROUTES"
app.conf.task_queues = (
    {
        "name": "io",
        "exchange": "io",
        "routing_key": "io",
    },
    {
        "name": "cpu",
        "exchange": "cpu",
        "routing_key": "cpu",
    },
)

# Pool configuration
pool_configs = {
    "prefork": "Default for CPU/most tasks",
    "gevent": "High-concurrency I/O (only if libs are compatible)",
    "eventlet": "Alternative to gevent for I/O",
    "solo": "Debugging and single-threaded execution"
}

# Worker configuration
worker_configs = {
    "concurrency": "≈ cores for CPU or higher for I/O",
    "autoscale": "Dynamic scaling based on queue length",
    "max_tasks_per_child": "Prevent memory leaks",
    "time_limit": "Hard kill for runaway tasks"
}

# Autoscaling settings
app.conf.update(
    worker_autoscaler="celery.worker.autoscale:Autoscaler",
    worker_max_tasks_per_child=1000,
    worker_max_memory_per_child=200000,  # 200MB
)

# app/task_composition.py
from celery import chain, group, chord
from .tasks import fetch, transform, store, notify

# Chain: sequential execution
workflow = chain(
    fetch.s("data_source"),
    transform.s(),
    store
)

# Group: parallel execution
parallel_tasks = group(
    fetch.s("source1"),
    fetch.s("source2"),
    fetch.s("source3")
)

# Chord: fan-in after parallel execution
chord_workflow = chord(
    group(fetch.s(i) for i in range(100)),
    transform.s()
)

# Complex workflow
complex_workflow = chain(
    group(fetch.s(i) for i in range(100)) |
    transform.s() |
    chord(store.s(), notify.s())
)

# Execute workflow
result = complex_workflow.apply_async()

# Check status
print(f"Workflow ID: {result.id}")
print(f"Status: {result.status}")

# Get results
if result.ready():
    print(f"Results: {result.get()}")

# app/beat.py
from . import celery
from celery.schedules import crontab

celery.conf.beat_schedule = {
    "nightly-report": {
        "task": "app.tasks.generate_report",
        "schedule": crontab(minute=0, hour=2),  # 02:00 UTC
        "options": {"queue": "cpu", "expires": 3600},
    },
    "cleanup-temp-files": {
        "task": "app.tasks.cleanup_temp_files",
        "schedule": crontab(minute=0, hour=1),  # 01:00 UTC
        "options": {"queue": "io", "expires": 1800},
    },
    "health-check": {
        "task": "app.tasks.health_check",
        "schedule": crontab(minute="*/5"),  # Every 5 minutes
        "options": {"queue": "io", "expires": 300},
    },
}

# Flower configuration
flower_config = {
    "port": 5555,
    "broker": "redis://localhost:6379/0",
    "basic_auth": ["admin:password"],  # Production auth
    "persistent": True,
    "db": "/tmp/flower.db"
}

# Structured logging configuration
import logging
import json

class CeleryFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": record.created,
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "task_id": getattr(record, "task_id", None),
            "task_name": getattr(record, "task_name", None),
        }
        return json.dumps(log_entry)

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    handlers=[logging.StreamHandler()],
    format="%(message)s"
)

# Prometheus metrics
from prometheus_client import Counter, Histogram, Gauge

# Metrics
task_counter = Counter('celery_tasks_total', 'Total tasks', ['task_name', 'status'])
task_duration = Histogram('celery_task_duration_seconds', 'Task duration', ['task_name'])
active_tasks = Gauge('celery_active_tasks', 'Active tasks', ['queue'])

# Task decorator for metrics
def track_metrics(task_func):
    def wrapper(*args, **kwargs):
        start_time = time.time()
        try:
            result = task_func(*args, **kwargs)
            task_counter.labels(task_name=task_func.__name__, status='success').inc()
            return result
        except Exception as exc:
            task_counter.labels(task_name=task_func.__name__, status='failure').inc()
            raise
        finally:
            task_duration.labels(task_name=task_func.__name__).observe(time.time() - start_time)
    return wrapper

# api/main.py
from fastapi import FastAPI, HTTPException
from app.tasks import send_webhook, generate_report

app = FastAPI()

@app.post("/submit")
def submit(item: dict):
    """Submit item for processing."""
    task = send_webhook.delay(
        "https://hooks.example.com",
        item,
        idem_key=item.get("id")
    )
    return {"queued": True, "task_id": task.id}

@app.get("/status/{task_id}")
def get_status(task_id: str):
    """Get task status."""
    task = send_webhook.AsyncResult(task_id)
    return {
        "task_id": task_id,
        "status": task.status,
        "result": task.result if task.ready() else None
    }

# 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"

# views.py
from django.http import JsonResponse
from .tasks import send_webhook

def submit_view(request):
    """Submit for processing."""
    task = send_webhook.delay(
        "https://hooks.example.com",
        request.POST.dict()
    )
    return JsonResponse({"queued": True, "task_id": task.id})

# Results backend configuration
app.conf.update(
    result_expires=3600,        # 1 hour TTL
    result_extended=True,        # Include task metadata
    result_persistent=True,      # Persist to disk
)

# Fire-and-forget pattern
@shared_task
def fire_and_forget_task(data: dict):
    """Task that doesn't need results."""
    process_data(data)
    # No return value needed

# Store artifacts externally
@shared_task
def generate_large_report(self, report_type: str) -> dict:
    """Generate report and store externally."""
    # Generate report
    report_data = build_report(report_type)

    # Store to S3
    report_url = save_to_s3(report_data)

    # Return only reference
    return {"report_url": report_url}

# Idempotent task with external key
@shared_task(bind=True, acks_late=True)
def idempotent_task(self, business_key: str, data: dict) -> dict:
    """Idempotent task with business key."""
    # Check if already processed
    existing = get_by_business_key(business_key)
    if existing:
        return {"status": "already_processed", "id": existing.id}

    # Process and store
    result = process_and_store(business_key, data)
    return {"status": "processed", "id": result.id}

# Retry with exponential backoff
@shared_task(
    bind=True,
    autoretry_for=(Exception,),
    retry_backoff=2,
    retry_jitter=True,
    retry_kwargs={"max_retries": 5}
)
def retryable_task(self, data: dict) -> dict:
    """Task with retry logic."""
    try:
        return process_data(data)
    except Exception as exc:
        # Log error and retry
        logger.error(f"Task failed: {exc}")
        self.retry(countdown=60, exc=exc)

# Saga pattern for multi-step workflows
@shared_task
def saga_step_1(data: dict) -> dict:
    """First step of saga."""
    result = process_step_1(data)
    return {"step_1_result": result, "data": data}

@shared_task
def saga_step_2(step_1_result: dict) -> dict:
    """Second step of saga."""
    try:
        result = process_step_2(step_1_result)
        return {"step_2_result": result}
    except Exception as exc:
        # Compensate for step 1
        compensate_step_1(step_1_result)
        raise

# Compensating action
@shared_task
def compensate_step_1(step_1_result: dict) -> None:
    """Compensate for step 1 failure."""
    rollback_step_1(step_1_result)

# Security settings
app.conf.update(
    broker_connection_retry_on_startup=True,
    broker_connection_retry=True,
    broker_connection_max_retries=10,
    broker_ssl_cert_reqs="required",
    broker_ssl_ca_certs="/path/to/ca.pem",
    broker_ssl_certfile="/path/to/cert.pem",
    broker_ssl_keyfile="/path/to/key.pem",
)

# Multi-tenant task
@shared_task
def tenant_aware_task(tenant_id: str, data: dict) -> dict:
    """Task with tenant isolation."""
    # Set tenant context
    set_tenant_context(tenant_id)

    # Process with tenant isolation
    result = process_with_tenant_isolation(data)

    return {"tenant_id": tenant_id, "result": result}

# test_tasks.py
import pytest
from unittest.mock import patch
from app.tasks import send_webhook

def test_send_webhook_success(mocker):
    """Test webhook task success."""
    mock_post = mocker.patch("httpx.Client.post")
    mock_post.return_value.status_code = 200

    result = send_webhook.run("http://example.com", {"test": "data"})

    assert result == {"status": 200}
    mock_post.assert_called_once()

def test_send_webhook_retry(mocker):
    """Test webhook task retry."""
    mock_post = mocker.patch("httpx.Client.post")
    mock_post.side_effect = httpx.HTTPError("Connection failed")

    with pytest.raises(httpx.HTTPError):
        send_webhook.run("http://example.com", {"test": "data"})

# test_integration.py
import pytest
from app.tasks import send_webhook

@pytest.fixture
def celery_app():
    """Test Celery app."""
    from app import make_celery
    app = make_celery()
    app.conf.update(
        task_always_eager=True,
        task_eager_propagates=True,
    )
    return app

def test_webhook_integration(celery_app):
    """Test webhook integration."""
    result = send_webhook.delay("http://example.com", {"test": "data"})
    assert result.status == "SUCCESS"

# DON'T: Giant tasks that run for hours
@shared_task
def bad_giant_task(data: list):
    """This will timeout and fail."""
    for item in data:  # Process 1M items
        process_item(item)  # This takes hours

# DO: Split into subtasks
@shared_task
def good_chunked_task(chunk: list):
    """Process chunk of data."""
    for item in chunk:
        process_item(item)

# Orchestrate with chord
chord(
    group(good_chunked_task.s(chunk) for chunk in chunks),
    finalize.s()
)

# DON'T: Non-idempotent side effects
@shared_task
def bad_charge_card(amount: int):
    """This will double-charge on retry."""
    charge_card(amount)  # No idempotency key!

# DO: Use idempotency key
@shared_task
def good_charge_card(amount: int, idem_key: str):
    """Idempotent card charging."""
    charge_card_with_key(amount, idem_key)

# DON'T: Mix CPU hogs with latency-sensitive tasks
# Same queue for both:
@shared_task
def cpu_intensive_task():
    """This will block I/O tasks."""
    return sum(range(1000000))

@shared_task
def latency_sensitive_task():
    """This needs fast response."""
    return send_notification()

# DO: Separate queues
# CPU queue: cpu_intensive_task
# I/O queue: latency_sensitive_task

# Essential Celery patterns
celery_patterns = {
    "use_celery_for": "HTTP offloading, CPU bursts, scheduled jobs, idempotent side-effects",
    "task_design": "Idempotent, retryable, bounded by time & rate",
    "queue_separation": "Separate queues for different task types",
    "worker_config": "Right pool, autoscale thoughtfully",
    "scheduling": "UTC schedules with Beat, expire late jobs",
    "observability": "Flower + metrics, configure DLQ + replayer",
    "results": "Keep results short-lived, store artifacts externally"
}

# Essential Celery operations
# 1. Create task
@shared_task(bind=True, acks_late=True)
def my_task(self, data):
    return process_data(data)

# 2. Execute task
result = my_task.delay(data)

# 3. Check status
print(result.status)

# 4. Get result
if result.ready():
    print(result.get())

# 5. Configure retries
@shared_task(autoretry_for=(Exception,), retry_kwargs={"max_retries": 3})
def retryable_task(data):
    return process_data(data)

# 6. Set time limits
@shared_task(time_limit=60, soft_time_limit=55)
def bounded_task(data):
    return process_data(data)