martinholovsky/claude-skills-generator/celery-expert

# 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

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

# 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

# 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

# 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

# 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

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

# 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

# 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

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())

from kombu import Exchange, Queue