Celery FastAPI

Automatic REST API generation for Celery tasks with FastAPI. This package seamlessly bridges Celery and FastAPI, automatically creating REST endpoints for all your registered Celery tasks.

Features

• 🚀 Automatic endpoint generation - REST APIs created automatically for all Celery tasks
• 🔧 Zero configuration - Works out of the box with sensible defaults
• 📊 Task monitoring - Built-in endpoints for task status, revocation, and worker info
• 🎯 App-scoped operations - Only manages tasks from your specific Celery app, not the entire cluster
• 🖥️ CLI support - Run as a standalone server from command line
• 📦 Modular design - Use as a library or standalone application
• 🔄 Queue-aware routing - Respects Celery queue assignments
• 📝 OpenAPI documentation - Full Swagger/ReDoc support
• 🔒 Production ready - Full uvicorn/gunicorn support with SSL, workers, and all options
• ⚡ Full Celery options - All task options (countdown, eta, priority, etc.)
• 🔌 Pool support - Compatible with eventlet, gevent, prefork, and solo pools
• 🧮 Batch execution - Submit groups of tasks in a single request via /tasks/batch
• 🛡️ Input validation - Pydantic-driven validation on task_name/queue at trust boundary
• 🔌 WebSocket streaming - Live task status updates via /tasks/{task_id}/ws

Requirements

• Python 3.11+
• FastAPI 0.100.0+
• Celery 5.3.0+

Installation

# Basic installation
pip install celery-fastapi

# With CLI support
pip install celery-fastapi[cli]

# With uvicorn server
pip install celery-fastapi[server]

# With gunicorn for production
pip install celery-fastapi[gunicorn]

# With Redis broker
pip install celery-fastapi[redis]

# With RabbitMQ broker
pip install celery-fastapi[rabbitmq]

# With eventlet/gevent concurrency
pip install celery-fastapi[eventlet]
pip install celery-fastapi[gevent]

# All extras (recommended for production)
pip install celery-fastapi[all]

Or with Poetry:

poetry add celery-fastapi
poetry add celery-fastapi --extras cli  # for CLI support

Quick Start

As a Python Module

from celery import Celery
from celery_fastapi import CeleryFastAPIBridge, create_app

# Your existing Celery app
celery_app = Celery('tasks', broker='redis://localhost:6379/0')

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

@celery_app.task
def multiply(x, y):
    return x * y

# Option 1: Using create_app factory
app = create_app(celery_app)

# Option 2: Using the Bridge class for more control
from fastapi import FastAPI

fastapi_app = FastAPI(title="My Task API")
bridge = CeleryFastAPIBridge(celery_app, fastapi_app)
bridge.register_routes()

Run with uvicorn:

uvicorn myapp:app --reload

Using the CLI

# Start the server (development)
celery-fastapi serve examples.celery_app:celery_app --port 8000 --reload

# Production with multiple workers
celery-fastapi serve examples.celery_app:celery_app -w 4 --host 0.0.0.0

# With custom worker hostname (for health checks)
export CELERY_WORKER_HOSTNAME="celery@worker1"
celery-fastapi serve examples.celery_app:celery_app --port 8000

# With SSL
celery-fastapi serve examples.celery_app:celery_app --ssl-keyfile key.pem --ssl-certfile cert.pem

# Using gunicorn (production)
celery-fastapi serve-gunicorn examples.celery_app:celery_app -w 4 -k uvicorn.workers.UvicornWorker

# List available routes
celery-fastapi routes examples.celery_app:celery_app

# List registered tasks
celery-fastapi tasks examples.celery_app:celery_app

# Show active workers
celery-fastapi workers examples.celery_app:celery_app

API Endpoints

Once running, your Celery tasks are available as REST endpoints:

Task Execution

# Execute a task with basic args
POST /{task_name_with_slashes}
Content-Type: application/json

{
    "args": [1, 2],
    "kwargs": {}
}

# Execute with advanced Celery options
POST /myapp/process_data
Content-Type: application/json

{
    "args": ["data.csv"],
    "kwargs": {"output_format": "json"},
    "countdown": 60,
    "priority": 5,
    "queue": "high_priority",
    "time_limit": 300,
    "soft_time_limit": 280
}

# Response
{
    "task_id": "abc123-def456-...",
    "status": "PENDING"
}

Task Status

# Get task status
GET /tasks/{task_id}

# Response
{
    "task_id": "abc123-def456-...",
    "state": "SUCCESS",
    "result": 3,
    "traceback": null,
    "date_done": "2024-01-15T10:30:00Z"
}

Task Management

# Revoke a task
POST /tasks/{task_id}/revoke
Content-Type: application/json

{
    "terminate": true,
    "signal": "SIGTERM"
}

# Get task result only
GET /tasks/{task_id}/result

# List active workers (filtered to this app's tasks)
GET /workers

# List available tasks in THIS app
GET /available-tasks

# Response
{
    "app_name": "my_tasks",
    "task_count": 4,
    "tasks": [
        {"name": "my_tasks.add", "queue": "default", ...},
        {"name": "my_tasks.multiply", "queue": "default", ...}
    ]
}

# List queues
GET /queues

# Purge tasks from a queue
POST /purge

Health Check and Monitoring

# Health check for local Celery worker
GET /healthz

# Response
{
    "status": "healthy",
    "celery_app": "example_tasks",
    "broker_connected": true,
    "worker_hostname": "celery@worker1",
    "worker_online": true
}

# Ping local Celery worker
GET /ping

# Response
{
    "worker_hostname": "celery@worker1",
    "online": true,
    "response": {"ok": "pong"}
}

Note: Health and ping endpoints automatically discover the local worker using:

1. CELERY_WORKER_HOSTNAME environment variable (recommended for custom hostnames)
2. Hostname matching (when worker and API share the same hostname)
3. Single worker fallback (when only one worker has this app's tasks)

For custom worker hostnames, set the environment variable:

export CELERY_WORKER_HOSTNAME="celery@worker1"
celery -A examples.celery_app worker --hostname worker1
celery-fastapi serve examples.celery_app:celery_app --port 8000

List All Tasks

# List active, scheduled, reserved, and revoked tasks (filtered to this app only)
GET /tasks

# Response
{
    "active": {...},
    "scheduled": {...},
    "reserved": {...},
    "revoked": {...}
}

Configuration

CeleryFastAPIBridge Options

bridge = CeleryFastAPIBridge(
    celery_app=celery_app,
    fastapi_app=fastapi_app,  # Optional, creates new if not provided
    prefix="/api/v1",         # URL prefix for all endpoints
    include_status_endpoints=True,  # Include /tasks endpoints
    task_filter=lambda name: not name.startswith("internal."),  # Filter tasks
)

create_app Options

app = create_app(
    celery_app,  # Celery instance or module path string
    title="My API",
    description="Task API",
    version="1.0.0",
    prefix="/api",
    include_status_endpoints=True,
    fastapi_kwargs={"docs_url": "/swagger"},
)

Integration with Existing FastAPI App

from fastapi import FastAPI
from celery_fastapi import CeleryFastAPIBridge
from myapp import celery_app

app = FastAPI()

# Your existing routes
@app.get("/health")
def health_check():
    return {"status": "healthy"}

# Add Celery task endpoints under /celery prefix
bridge = CeleryFastAPIBridge(
    celery_app,
    app,
    prefix="/celery",
)
bridge.register_routes()

CLI Reference

celery-fastapi --help

Commands:
  serve            Start the FastAPI server with uvicorn
  serve-gunicorn   Start the FastAPI server with Gunicorn
  routes           List all generated routes
  tasks            List all registered Celery tasks
  workers          Show active Celery workers

# Serve options (uvicorn)
celery-fastapi serve examples.celery_app:celery_app \
    --host 0.0.0.0 \
    --port 8000 \
    --reload \
    --workers 4 \
    --prefix /api \
    --log-level info \
    --ssl-keyfile key.pem \
    --ssl-certfile cert.pem \
    --proxy-headers \
    --forwarded-allow-ips '*'

# Serve options (gunicorn)
celery-fastapi serve-gunicorn examples.celery_app:celery_app \
    --bind 0.0.0.0:8000 \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --timeout 30 \
    --daemon \
    --pid /var/run/celery-fastapi.pid

Development

# Clone the repository
git clone https://github.com/karailker/celery-fastapi.git
cd celery-fastapi

# Install dependencies
poetry install --extras all

# Run tests
poetry run pytest

# Run linting
poetry run ruff check .
poetry run mypy celery_fastapi

# Format code
poetry run ruff format .

License

MIT License - see LICENSE file for details.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

1. Fork the repository
2. Create your feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request