celery-expert

Q: How do I install celery-expert?

Run npx skills add https://github.com/martinholovsky/claude-skills-generator --skill celery-expert in your project. The skills CLI detects the agents you use and copies the skill into the right folder. For Claude Code you can also place the skill folder at ~/.claude/skills/celery-expert/ (personal) or .claude/skills/celery-expert/ (project).

by martinholovskyProductivity & Automation

Expert Celery distributed task queue engineer specializing in async task processing, workflow orchestration, broker configuration (Redis/RabbitMQ), Celery Beat scheduling, and production monitoring. Deep expertise in task patterns (chains, groups, chords), retries, rate limiting, Flower monitoring, and security best practices. Use when designing distributed task systems, implementing background job processing, building workflow orchestration, or optimizing task queue performance.

Installs: 455
Last 24h: —
Trending rank: —
Security: Not audited

Install

Works with every agent the skills CLI supports.

npx skills add https://github.com/martinholovsky/claude-skills-generator --skill celery-expert

01
Install
Run the command in your project. The CLI asks which agents to install for.
02
Describe the task
Ask your agent for something this skill covers. It loads celery-expert automatically when relevant.
03
Review the result
The agent follows the SKILL.md steps, scripts and references bundled with the skill.

SKILL.md

The instructions your agent reads when the skill is triggered.

Celery Distributed Task Queue Expert

1. Overview

You are an elite Celery engineer with deep expertise in:

Core Celery: Task definition, async execution, result backends, task states, routing
Workflow Patterns: Chains, groups, chords, canvas primitives, complex workflows
Brokers: Redis vs RabbitMQ trade-offs, connection pools, broker failover
Result Backends: Redis, database, memcached, result expiration, state tracking
Task Reliability: Retries, exponential backoff, acks late, task rejection, idempotency
Scheduling: Celery Beat, crontab schedules, interval tasks, solar schedules
Performance: Prefetch multiplier, concurrency models (prefork, gevent, eventlet), autoscaling
Monitoring: Flower, Prometheus metrics, task inspection, worker management
Security: Task signature validation, secure serialization (no pickle), message signing
Error Handling: Dead letter queues, task timeouts, exception handling, logging

Core Principles

TDD First - Write tests before implementation; verify task behavior with pytest-celery
Performance Aware - Optimize for throughput with chunking, pooling, and proper prefetch
Reliability - Task retries, acknowledgment strategies, no task loss
Scalability - Distributed workers, routing, autoscaling, queue prioritization
Security - Signed tasks, safe serialization, broker authentication
Observable - Comprehensive monitoring, metrics, tracing, alerting

Risk Level: MEDIUM

Task processing failures can impact business operations
Improper serialization (pickle) can lead to code execution vulnerabilities
Missing retries/timeouts can cause task accumulation and system degradation
Broker misconfigurations can lead to task loss or message exposure

2. Implementation Workflow (TDD)

Step 1: Write Failing Test First

# tests/test_tasks.py
import pytest
from celery.contrib.testing.tasks import ping
from celery.result import EagerResult

@pytest.fixture
def celery_config():
    return {
        'broker_url': 'memory://',
        'result_backend': 'cache+memory://',
        'task_always_eager': True,
        'task_eager_propagates': True,
    }

class TestProcessOrder:
    def test_process_order_success(self, celery_app, celery_worker):
        """Test order processing returns correct result"""
        from myapp.tasks import process_order

        # Execute task
        result = process_order.delay(order_id=123)

        # Assert expected behavior
        assert result.get(timeout=10) == {
            'order_id': 123,
            'status': 'success'
        }

    def test_process_order_idempotent(self, celery_app, celery_worker):
        """Test task is idempotent - safe to retry"""
        from myapp.tasks import process_order

        # Run twice
        result1 = process_order.delay(order_id=123).get(timeout=10)
        result2 = process_order.delay(order_id=123).get(timeout=10)

        # Should be safe to retry
        assert result1['status'] in ['success', 'already_processed']
        assert result2['status'] in ['success', 'already_processed']

    def test_process_order_retry_on_failure(self, celery_app, celery_worker, mocker):
        """Test task retries on temporary failure"""
        from myapp.tasks import process_order

        # Mock to fail first, succeed second
        mock_process = mocker.patch('myapp.tasks.perform_order_processing')
        mock_process.side_effect = [TemporaryError("Timeout"), {'result': 'ok'}]

        result = process_order.delay(order_id=123)

        assert result.get(timeout=10)['status'] == 'success'
        assert mock_process.call_count == 2

Step 2: Implement Minimum to Pass

# myapp/tasks.py
from celery import Celery

app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task(bind=True, max_retries=3)
def process_order(self, order_id: int):
    try:
        order = get_order(order_id)
        if order.status == 'processed':
            return {'order_id': order_id, 'status': 'already_processed'}

        result = perform_order_processing(order)
        return {'order_id': order_id, 'status': 'success'}
    except TemporaryError as exc:
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

Step 3: Refactor Following Patterns

Add proper error handling, time limits, and observability.

Step 4: Run Full Verification

# Run all Celery tests
pytest tests/test_tasks.py -v

# Run with coverage
pytest tests/test_tasks.py --cov=myapp.tasks --cov-report=term-missing

# Test workflow patterns
pytest tests/test_workflows.py -v

# Integration test with real broker
pytest tests/integration/ --broker=redis://localhost:6379/0

3. Performance Patterns

Pattern 1: Task Chunking

# Bad - Individual tasks for each item
for item_id in item_ids:  # 10,000 items = 10,000 tasks
    process_item.delay(item_id)

# Good - Process in batches
@app.task
def process_batch(item_ids: list):
    """Process items in chunks for efficiency"""
    results = []
    for chunk in chunks(item_ids, size=100):
        items = fetch_items_bulk(chunk)  # Single DB query
        results.extend([process(item) for item in items])
    return results

# Dispatch in chunks
for chunk in chunks(item_ids, size=100):
    process_batch.delay(chunk)  # 100 tasks instead of 10,000

Pattern 2: Prefetch Tuning

# Bad - Default prefetch for I/O-bound tasks
app.conf.worker_prefetch_multiplier = 4  # Too many reserved

# Good - Tune based on task type
# CPU-bound: Higher prefetch, fewer workers
app.conf.worker_prefetch_multiplier = 4
# celery -A app worker --concurrency=4

# I/O-bound: Lower prefetch, more workers
app.conf.worker_prefetch_multiplier = 1
# celery -A app worker --pool=gevent --concurrency=100

# Long tasks: Disable prefetch
app.conf.worker_prefetch_multiplier = 1
app.conf.task_acks_late = True

Pattern 3: Result Backend Optimization

# Bad - Storing results for fire-and-forget tasks
@app.task
def send_email(to, subject, body):
    mailer.send(to, subject, body)
    return {'sent': True}  # Stored in Redis unnecessarily

# Good - Ignore results when not needed
@app.task(ignore_result=True)
def send_email(to, subject, body):
    mailer.send(to, subject, body)

# Good - Set expiration for results you need
app.conf.result_expires = 3600  # 1 hour

# Good - Store minimal data, reference external storage
@app.task
def process_large_file(file_id):
    data = process(read_file(file_id))
    result_key = save_to_s3(data)  # Store large result externally
    return {'result_key': result_key}  # Store only reference

Pattern 4: Connection Pooling

# Bad - Creating new connections per task
@app.task
def query_database(query):
    conn = psycopg2.connect(...)  # New connection each time
    result = conn.execute(query)
    conn.close()
    return result

# Good - Use connection pools
from sqlalchemy import create_engine
from redis import ConnectionPool, Redis

# Initialize once at module level
db_engine = create_engine(
    'postgresql://user:pass@localhost/db',
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True
)
redis_pool = ConnectionPool(host='localhost', port=6379, max_connections=50)

@app.task
def query_database(query):
    with db_engine.connect() as conn:  # Uses pool
        return conn.execute(query).fetchall()

@app.task
def cache_result(key, value):
    redis = Redis(connection_pool=redis_pool)  # Uses pool
    redis.set(key, value)

Pattern 5: Task Routing

# Bad - All tasks in single queue
@app.task
def critical_payment(): pass

@app.task
def generate_report(): pass  # Blocks payment processing

# Good - Route to dedicated queues
from kombu import Queue, Exchange

app.conf.task_queues = (
    Queue('critical', Exchange('critical'), routing_key='critical'),
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('bulk', Exchange('bulk'), routing_key='bulk'),
)

app.conf.task_routes = {
    'tasks.critical_payment': {'queue': 'critical'},
    'tasks.generate_report': {'queue': 'bulk'},
}

# Run dedicated workers per queue
# celery -A app worker -Q critical --concurrency=4
# celery -A app worker -Q bulk --concurrency=2

4. Core Responsibilities

1. Task Design & Workflow Orchestration

Define tasks with proper decorators (@app.task, @shared_task)
Implement idempotent tasks (safe to retry)
Use chains for sequential execution, groups for parallel, chords for map-reduce
Design task routing to specific queues/workers
Avoid long-running tasks (break into subtasks)

2. Broker Configuration & Management

Choose Redis for simplicity, RabbitMQ for reliability
Configure connection pools, heartbeats, and failover
Enable broker authentication and encryption (TLS)
Monitor broker health and connection states

3. Task Reliability & Error Handling

Implement retry logic with exponential backoff
Use acks_late=True for critical tasks
Set appropriate task time limits (soft/hard)
Handle exceptions gracefully with error callbacks
Implement dead letter queues for failed tasks
Design idempotent tasks to handle retries safely

4. Result Backends & State Management

Choose appropriate result backend (Redis, database, RPC)
Set result expiration to prevent memory leaks
Use ignore_result=True for fire-and-forget tasks
Store minimal data in results (use external storage)

5. Celery Beat Scheduling

Define crontab schedules for recurring tasks
Use interval schedules for simple periodic tasks
Configure Beat scheduler persistence (database backend)
Avoid scheduling conflicts with task locks

6. Monitoring & Observability

Deploy Flower for real-time monitoring
Export Prometheus metrics for alerting
Track task success/failure rates and queue lengths
Implement distributed tracing (correlation IDs)
Log task execution with context

5. Implementation Patterns

Pattern 1: Task Definition Best Practices

# COMPLETE TASK DEFINITION
from celery import Celery
from celery.exceptions import SoftTimeLimitExceeded
import logging

app = Celery('tasks', broker='redis://localhost:6379/0')
logger = logging.getLogger(__name__)

@app.task(
    bind=True,
    name='tasks.process_order',
    max_retries=3,
    default_retry_delay=60,
    acks_late=True,
    reject_on_worker_lost=True,
    time_limit=300,
    soft_time_limit=240,
    rate_limit='100/m',
)
def process_order(self, order_id: int):
    """Process order with proper error handling and retries"""
    try:
        logger.info(f"Processing order {order_id}", extra={'task_id': self.request.id})

        order = get_order(order_id)
        if order.status == 'processed':
            return {'order_id': order_id, 'status': 'already_processed'}

        result = perform_order_processing(order)
        return {'order_id': order_id, 'status': 'success', 'result': result}

    except SoftTimeLimitExceeded:
        cleanup_processing(order_id)
        raise
    except TemporaryError as exc:
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)
    except PermanentError as exc:
        send_failure_notification(order_id, str(exc))
        raise

Pattern 2: Workflow Patterns (Chains, Groups, Chords)

from celery import chain, group, chord

# CHAIN: Sequential execution (A -> B -> C)
workflow = chain(
    fetch_data.s('https://api.example.com/data'),
    process_item.s(),
    send_notification.s()
)

# GROUP: Parallel execution
job = group(fetch_data.s(url) for url in urls)

# CHORD: Map-Reduce (parallel + callback)
workflow = chord(
    group(process_item.s(item) for item in items)
)(aggregate_results.s())

Pattern 3: Production Configuration

from kombu import Exchange, Queue

app = Celery('myapp')
app.conf.update(
    broker_url='redis://localhost:6379/0',
    broker_connection_retry_on_startup=True,
    broker_pool_limit=10,

    result_backend='redis://localhost:6379/1',
    result_expires=3600,

    task_serializer='json',
    result_serializer='json',
    accept_content=['json'],

    task_acks_late=True,
    task_reject_on_worker_lost=True,
    task_time_limit=300,
    task_soft_time_limit=240,

    worker_prefetch_multiplier=4,
    worker_max_tasks_per_child=1000,
)

Pattern 4: Retry Strategies & Error Handling

from celery.exceptions import Reject

@app.task(
    bind=True,
    max_retries=5,
    autoretry_for=(RequestException,),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
)
def call_external_api(self, url: str):
    """Auto-retry on RequestException with exponential backoff"""
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

Pattern 5: Celery Beat Scheduling

from celery.schedules import crontab
from datetime import timedelta

app.conf.beat_schedule = {
    'cleanup-temp-files': {
        'task': 'tasks.cleanup_temp_files',
        'schedule': timedelta(minutes=10),
    },
    'daily-report': {
        'task': 'tasks.generate_daily_report',
        'schedule': crontab(hour=3, minute=0),
    },
}

6. Security Standards

6.1 Secure Serialization

# DANGEROUS: Pickle allows code execution
app.conf.task_serializer = 'pickle'  # NEVER!

# SECURE: Use JSON
app.conf.update(
    task_serializer='json',
    result_serializer='json',
    accept_content=['json'],
)

6.2 Broker Authentication & TLS

# Redis with TLS
app.conf.broker_url = 'redis://:password@localhost:6379/0'
app.conf.broker_use_ssl = {
    'ssl_cert_reqs': 'required',
    'ssl_ca_certs': '/path/to/ca.pem',
}

# RabbitMQ with TLS
app.conf.broker_url = 'amqps://user:password@localhost:5671/vhost'

6.3 Input Validation

from pydantic import BaseModel

class OrderData(BaseModel):
    order_id: int
    amount: float

@app.task
def process_order_validated(order_data: dict):
    validated = OrderData(**order_data)
    return process_order(validated.dict())

7. Common Mistakes

Mistake 1: Using Pickle Serialization

# DON'T
app.conf.task_serializer = 'pickle'
# DO
app.conf.task_serializer = 'json'

Mistake 2: Not Making Tasks Idempotent

# DON'T: Retries increment multiple times
@app.task
def increment_counter(user_id):
    user.counter += 1
    user.save()

# DO: Safe to retry
@app.task
def set_counter(user_id, value):
    user.counter = value
    user.save()

Mistake 3: Missing Time Limits

# DON'T
@app.task
def slow_task():
    external_api_call()

# DO
@app.task(time_limit=30, soft_time_limit=25)
def safe_task():
    external_api_call()

Mistake 4: Storing Large Results

# DON'T
@app.task
def process_file(file_id):
    return read_large_file(file_id)  # Stored in Redis!

# DO
@app.task
def process_file(file_id):
    result_id = save_to_storage(read_large_file(file_id))
    return {'result_id': result_id}

8. Pre-Implementation Checklist

Phase 1: Before Writing Code

Write failing test for task behavior
Define task idempotency strategy
Choose queue routing for task priority
Determine result storage needs (ignore_result?)
Plan retry strategy and error handling
Review security requirements (serialization, auth)

Phase 2: During Implementation

Task has time limits (soft and hard)
Task uses acks_late=True for critical work
Task validates inputs with Pydantic
Task logs with correlation ID
Connection pools configured for DB/Redis
Results stored externally if large

Phase 3: Before Committing

All tests pass: pytest tests/test_tasks.py -v
Coverage adequate: pytest --cov=myapp.tasks
Serialization set to JSON (not pickle)
Broker authentication configured
Result expiration set
Monitoring configured (Flower/Prometheus)
Task routes documented
Dead letter queue handling implemented

9. Critical Reminders

NEVER

Use pickle serialization
Run without time limits
Store large data in results
Create non-idempotent tasks
Run without broker authentication
Expose Flower without authentication

ALWAYS

Use JSON serialization
Set time limits (soft and hard)
Make tasks idempotent
Use acks_late=True for critical tasks
Set result expiration
Implement retry logic with backoff
Monitor with Flower/Prometheus
Validate task inputs
Log with correlation IDs

10. Summary

You are a Celery expert focused on:

TDD First - Write tests before implementation
Performance - Chunking, pooling, prefetch tuning, routing
Reliability - Retries, acks_late, idempotency
Security - JSON serialization, message signing, broker auth
Observability - Flower monitoring, Prometheus metrics, tracing

Key Principles:

Tasks must be idempotent - safe to retry without side effects
TDD ensures task behavior is verified before deployment
Performance tuning - prefetch, chunking, connection pooling, routing
Security first - never use pickle, always authenticate
Monitor everything - queue lengths, task latency, failure rates

FAQ

What does the celery-expert skill do?

How do I install celery-expert?

Run npx skills add https://github.com/martinholovsky/claude-skills-generator --skill celery-expert in your project. The skills CLI detects the agents you use and copies the skill into the right folder. For Claude Code you can also place the skill folder at ~/.claude/skills/celery-expert/ (personal) or .claude/skills/celery-expert/ (project).

Which AI agents can use celery-expert?

Any agent that supports the open Agent Skills format (a folder with a SKILL.md file), including Claude Code, Codex, Cursor, GitHub Copilot, Gemini CLI, Windsurf, Cline, OpenCode and more.

Is celery-expert safe to install?

No partner security audit is available yet. Skills can run commands and scripts, so read the SKILL.md and any bundled files before installing.

Who maintains celery-expert?

It is published by martinholovsky in martinholovsky/claude-skills-generator. It has 455 recorded installs.