celery-config-patterns

Provides production-ready Celery configuration templates for all major Python frameworks (Django, Flask, FastAPI, standalone) with complete broker setup (Redis, RabbitMQ), security best practices, and framework-specific integration patterns.

Use When

• Configuring Celery for the first time in any Python project

• Integrating Celery with Django, Flask, FastAPI, or standalone applications

• Setting up Redis or RabbitMQ as message broker

• Creating production-ready Celery configurations

• Implementing task routing and queue configurations

• Setting up monitoring and logging for Celery

• Migrating Celery configurations between frameworks

Directory Structure

celery-config-patterns/ ├── SKILL.md # This file ├── scripts/ │ ├── validate-config.sh # Validate Celery configuration │ ├── detect-framework.sh # Auto-detect Python framework │ ├── generate-config.sh # Generate framework-specific config │ └── test-broker-connection.sh # Test broker connectivity ├── templates/ │ ├── celery-app-standalone.py # Standalone Celery app │ ├── celery-app-django.py # Django Celery integration │ ├── celery-app-flask.py # Flask Celery integration │ ├── celery-app-fastapi.py # FastAPI Celery integration │ ├── config-redis.py # Redis broker configuration │ ├── config-rabbitmq.py # RabbitMQ broker configuration │ ├── tasks-example.py # Sample task definitions │ └── beat-schedule.py # Celery Beat schedule config └── examples/ ├── django-setup.md # Complete Django setup guide ├── flask-setup.md # Complete Flask setup guide ├── fastapi-setup.md # Complete FastAPI setup guide └── standalone-setup.md # Standalone Python setup guide

Instructions

Step 1: Detect Framework

Use the detection script to identify the Python framework:

bash scripts/detect-framework.sh

Detects:

• Django (looks for manage.py , settings.py )

• Flask (looks for Flask imports, app.py )

• FastAPI (looks for FastAPI imports, main.py )

• Standalone (no framework detected)

Step 2: Select Template

Based on framework detection, choose the appropriate template:

Framework Template Integration File

Django celery-app-django.py

projectname/celery.py

Flask celery-app-flask.py

app/celery.py or celery.py

FastAPI celery-app-fastapi.py

app/celery.py or celery.py

Standalone celery-app-standalone.py

celery_app.py

Step 3: Configure Broker

Choose broker configuration template:

Redis (Recommended for simplicity):

Use templates/config-redis.py

CELERY_BROKER_URL = 'redis://localhost:6379/0' CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

RabbitMQ (Recommended for production):

Use templates/config-rabbitmq.py

CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672//' CELERY_RESULT_BACKEND = 'rpc://'

Step 4: Generate Configuration

Use the generation script:

bash scripts/generate-config.sh --framework=django --broker=redis bash scripts/generate-config.sh --framework=flask --broker=rabbitmq bash scripts/generate-config.sh --framework=fastapi --broker=redis

Script will:

• Copy appropriate framework template

• Apply broker configuration

• Create environment file with placeholders

• Add to .gitignore if needed

• Generate setup documentation

Step 5: Validate Configuration

Run validation script before starting Celery:

bash scripts/validate-config.sh

Validates:

• Celery app exists and is importable

• Broker connection is valid

• Configuration syntax is correct

• Required environment variables are set

• Task discovery paths are correct

Step 6: Test Broker Connection

Verify broker connectivity:

bash scripts/test-broker-connection.sh

Tests:

• Broker URL is reachable

• Authentication credentials are valid

• Connection pool can be established

• Basic message routing works

Template Descriptions

celery-app-standalone.py

Standalone Celery application without web framework integration.

Features:

• Basic Celery app configuration

• Task autodiscovery from tasks.py

• Configurable broker and backend

• Logging setup

• Beat schedule integration

Usage:

celery_app.py

from celery import Celery

app = Celery('myapp') app.config_from_object('celeryconfig')

@app.task def add(x, y): return x + y

celery-app-django.py

Django-specific Celery configuration using Django settings.

Features:

• Reads configuration from Django settings

• Auto-discovers tasks in all installed apps

• Uses Django's database for result backend (optional)

• Integrates with Django logging

• Supports Django-celery-beat for periodic tasks

Usage:

myproject/celery.py

import os from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings') app = Celery('myproject') app.config_from_object('django.conf:settings', namespace='CELERY') app.autodiscover_tasks()

celery-app-flask.py

Flask-specific Celery configuration with Flask app context.

Features:

• Flask app factory pattern support

• Celery app context integration

• Uses Flask configuration

• Compatible with Flask-Celery extensions

• Request context handling for tasks

Usage:

celery.py

from celery import Celery

def make_celery(app): celery = Celery(app.import_name) celery.conf.update(app.config) return celery

app.py

celery = make_celery(app)

celery-app-fastapi.py

FastAPI-specific Celery configuration with async support.

Features:

• FastAPI lifespan event integration

• Async task support

• Pydantic model validation in tasks

• Background task coordination

• API endpoint integration examples

Usage:

celery.py

from celery import Celery

celery_app = Celery('fastapi_app') celery_app.config_from_object('celeryconfig')

main.py

from fastapi import FastAPI from celery_app import celery_app

config-redis.py

Redis broker and result backend configuration.

Features:

• Connection pooling

• SSL/TLS support

• Sentinel configuration

• Health checks

• Connection retry logic

Configuration:

CELERY_BROKER_URL = 'redis://localhost:6379/0' CELERY_RESULT_BACKEND = 'redis://localhost:6379/0' CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP = True CELERY_REDIS_MAX_CONNECTIONS = 50

config-rabbitmq.py

RabbitMQ broker configuration with advanced routing.

Features:

• Virtual host configuration

• Exchange and queue declarations

• Routing keys and bindings

• Dead letter queues

• Priority queues

Configuration:

CELERY_BROKER_URL = 'amqp://user:pass@localhost:5672//' CELERY_RESULT_BACKEND = 'rpc://' CELERY_TASK_ROUTES = { 'app.tasks.critical': {'queue': 'critical'}, 'app.tasks.normal': {'queue': 'default'}, }

tasks-example.py

Example task definitions with best practices.

Features:

• Basic task examples

• Task with retry logic

• Task with rate limiting

• Task with time limits

• Task with custom routing

• Task with result expiration

Examples:

@app.task(bind=True, max_retries=3) def process_data(self, data): try: # Process data return result except Exception as exc: raise self.retry(exc=exc, countdown=60)

@app.task(rate_limit='10/m') def send_email(to, subject, body): # Send email pass

beat-schedule.py

Celery Beat periodic task scheduling.

Features:

• Crontab schedules

• Interval schedules

• Solar schedules

• Task arguments and kwargs

• Timezone support

Examples:

CELERY_BEAT_SCHEDULE = { 'cleanup-every-midnight': { 'task': 'app.tasks.cleanup', 'schedule': crontab(hour=0, minute=0), }, 'report-every-monday': { 'task': 'app.tasks.weekly_report', 'schedule': crontab(day_of_week=1, hour=9), }, }

Script Usage

validate-config.sh

Validates Celery configuration for errors.

Usage:

bash scripts/validate-config.sh bash scripts/validate-config.sh --config=celeryconfig.py bash scripts/validate-config.sh --app=myapp.celery:app

Checks:

• Python syntax is valid

• Celery app is importable

• Broker URL format is correct

• Required settings are present

• Task modules can be discovered

• No conflicting configurations

Exit Codes:

• 0: Configuration is valid

• 1: Validation errors found

• 2: Import errors

detect-framework.sh

Detects Python web framework in current project.

Usage:

bash scripts/detect-framework.sh bash scripts/detect-framework.sh /path/to/project

Output:

{ "framework": "django", "version": "4.2.0", "celery_location": "myproject/celery.py", "settings_location": "myproject/settings.py" }

Detection Logic:

• Django: Check for manage.py and settings.py

• Flask: Check for Flask imports and app.py

• FastAPI: Check for FastAPI imports and main.py

• Standalone: No framework files detected

generate-config.sh

Generates framework-specific Celery configuration.

Usage:

bash scripts/generate-config.sh --framework=django --broker=redis bash scripts/generate-config.sh --framework=flask --broker=rabbitmq --output=celery_config.py

Options:

• --framework : django, flask, fastapi, standalone

• --broker : redis, rabbitmq

• --output : Custom output file path

• --with-beat : Include Celery Beat configuration

• --with-monitoring : Include monitoring configuration

Generated Files:

• Celery app configuration

• .env.example with broker credentials

• Setup documentation

• Example tasks file

test-broker-connection.sh

Tests message broker connectivity.

Usage:

bash scripts/test-broker-connection.sh bash scripts/test-broker-connection.sh redis://localhost:6379/0 bash scripts/test-broker-connection.sh amqp://guest:guest@localhost:5672//

Tests:

• DNS resolution of broker host

• Port connectivity

• Authentication

• Basic message publish/consume

• Connection pooling

Output:

✓ Broker host is reachable ✓ Port 6379 is open ✓ Authentication successful ✓ Message publish successful ✓ Message consume successful ✓ Connection pool working

All tests passed!

Examples

Django Setup Example

See examples/django-setup.md for complete walkthrough:

• Install dependencies

• Create Celery app in myproject/celery.py

• Configure in settings.py

• Create tasks in app directories

• Run worker and beat

Flask Setup Example

See examples/flask-setup.md for complete walkthrough:

• Install dependencies

• Create Celery factory function

• Initialize with Flask app

• Define tasks

• Run worker

FastAPI Setup Example

See examples/fastapi-setup.md for complete walkthrough:

• Install dependencies

• Create Celery app

• Integrate with FastAPI lifespan

• Create async tasks

• Run worker alongside FastAPI

Standalone Setup Example

See examples/standalone-setup.md for complete walkthrough:

• Install Celery and broker

• Create celery_app.py

• Create celeryconfig.py

• Define tasks

• Run worker

Security Requirements

All configuration templates and examples in this skill follow strict security rules:

Environment Variables:

• Broker credentials use placeholders: your_redis_password_here

• Connection strings never include real passwords

• All sensitive values read from environment

Example .env.example:

Redis Configuration

CELERY_BROKER_URL=redis://:your_redis_password_here@localhost:6379/0 CELERY_RESULT_BACKEND=redis://:your_redis_password_here@localhost:6379/0

RabbitMQ Configuration

CELERY_BROKER_URL=amqp://your_rabbitmq_user:your_rabbitmq_password@localhost:5672//

Code Examples:

ALWAYS read from environment

import os CELERY_BROKER_URL = os.getenv('CELERY_BROKER_URL', 'redis://localhost:6379/0')

Security Checklist:

• No hardcoded broker passwords

• No hardcoded database credentials

• All examples use environment variables

• .env added to .gitignore

• .env.example provided with placeholders

• Documentation explains credential management

Configuration Best Practices

Production Settings:

Task execution

CELERY_TASK_SERIALIZER = 'json' CELERY_RESULT_SERIALIZER = 'json' CELERY_ACCEPT_CONTENT = ['json'] CELERY_TIMEZONE = 'UTC' CELERY_ENABLE_UTC = True

Performance

CELERY_WORKER_PREFETCH_MULTIPLIER = 4 CELERY_WORKER_MAX_TASKS_PER_CHILD = 1000 CELERY_BROKER_POOL_LIMIT = 10

Reliability

CELERY_TASK_ACKS_LATE = True CELERY_TASK_REJECT_ON_WORKER_LOST = True CELERY_TASK_TIME_LIMIT = 300 CELERY_TASK_SOFT_TIME_LIMIT = 240

Development Settings:

Easier debugging

CELERY_TASK_ALWAYS_EAGER = True CELERY_TASK_EAGER_PROPAGATES = True

Troubleshooting

Broker Connection Failed:

Test broker connectivity

bash scripts/test-broker-connection.sh

Check broker is running

Redis: redis-cli ping

RabbitMQ: rabbitmqctl status

Tasks Not Discovered:

Validate configuration

bash scripts/validate-config.sh

Check task module paths

celery -A myapp inspect registered

Import Errors:

Verify Python path includes project

export PYTHONPATH="${PYTHONPATH}:/path/to/project"

Check Celery app is importable

python -c "from myapp.celery import app; print(app)"

Requirements

Python Packages:

• celery>=5.3.0

• redis>=4.5.0 (for Redis broker)

• kombu>=5.3.0 (for RabbitMQ broker)

Framework-Specific:

• Django: django-celery-beat, django-celery-results

• Flask: Flask-Celery

• FastAPI: No additional packages

Broker Requirements:

• Redis: redis-server running on port 6379

• RabbitMQ: rabbitmq-server running on port 5672

System Tools:

• bash (for scripts)

• python3 (3.8+)

• Access to broker (Redis/RabbitMQ)

Skill Version: 1.0.0 Last Updated: 2025-11-16