Using Python Celery with FastAPI and Django

python celery with fastapi and django requires a clear understanding of the core syntax, runtime behavior, and practical implementation patterns demonstrated in the examples below.

Celery is a distributed task queue that fits both Django and FastAPI because both are Python web frameworks with synchronous request handlers that should not block on long-running work. If you maintain a project that uses both frameworks—for example, a Django admin site alongside a FastAPI API—you can run a single Celery cluster and share task definitions between them.

Setting Up Celery in Django

Django integrates with Celery through a dedicated celery.py module in your project package. The standard setup defines a Celery application instance and loads configuration from Django settings.

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

In myproject/__init__.py, import the app so it is loaded when Django starts:

# myproject/__init__.py
from .celery import app as celery_app

__all__ = ('celery_app',)

Then configure the broker and result backend in settings.py:

# settings.py
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'

This setup allows Django apps to define tasks using the @shared_task decorator, which creates a task bound to the current Celery app without requiring a direct import of the app instance.

Setting Up Celery in FastAPI

FastAPI does not have a built-in integration like Django's, but you can create a Celery instance in a module and import it wherever needed. A common pattern is to define the Celery app in a separate file, such as celery_app.py, and use FastAPI's lifespan to manage the connection.

# celery_app.py
from celery import Celery

celery_app = Celery('fastapi_app', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')
celery_app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
)

In your FastAPI application, you can import this instance directly. There is no need to tie it to the ASGI lifespan unless you need to clean up resources on shutdown. For most use cases, importing the Celery app and using it to send tasks is sufficient.

# main.py
from fastapi import FastAPI
from celery_app import celery_app

app = FastAPI()

@app.post("/send-email")
async def send_email(email: str):
    celery_app.send_task('tasks.send_email', args=[email])
    return {"status": "queued"}

Because FastAPI runs on an async event loop, you should never call a Celery task synchronously inside a async def endpoint. Instead, use send_task or .delay() to enqueue the task and return immediately.

Defining Tasks That Work in Both Frameworks

To share tasks between Django and FastAPI, define them in a module that both applications can import. The @shared_task decorator from Celery is ideal because it does not bind the task to a specific Celery app instance.

# tasks.py
from celery import shared_task

@shared_task
def send_email(to_address: str, subject: str, body: str):
    # Simulate sending an email
    print(f"Sending email to {to_address}: {subject}")
    return {"to": to_address, "subject": subject}

In Django, you can place this file inside a Django app and use app.autodiscover_tasks() to find it. In FastAPI, you can import the task directly and call it with .delay():

from tasks import send_email

send_email.delay('user@example.com', 'Hello', 'Body')

The task module must be importable by both the Django and FastAPI processes. If your project is structured as a monorepo with separate packages, ensure the module is on PYTHONPATH for both.

Calling Tasks from Django Views and FastAPI Endpoints

Django views are synchronous, so you can call a Celery task directly using .delay() or .apply_async(). For example, in a Django view:

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

def notify_user(request):
    send_email.delay(request.user.email, "Welcome", "Thanks for signing up")
    return JsonResponse({"status": "queued"})

FastAPI endpoints can be either def (sync) or async def. In a sync endpoint, you can call .delay() directly. In an async endpoint, you should still call .delay() because it is a non-blocking operation that returns immediately; the Celery client does not block the event loop.

# main.py
from fastapi import FastAPI
from tasks import send_email

app = FastAPI()

@app.post("/notify")
async def notify(email: str):
    send_email.delay(email, "Notification", "You have a new message")
    return {"status": "queued"}

Both frameworks enqueue tasks in the same way, so you can reuse the same task definitions without duplication.

Handling Task Results and Errors

Celery can store task results in a result backend. To retrieve a result, you use the AsyncResult object. In Django, you can do this inside a view, but be careful not to block the request for a long-running task. Typically, you poll the result status or use a callback.

from celery.result import AsyncResult

def get_task_status(request, task_id):
    result = AsyncResult(task_id)
    return JsonResponse({'state': result.state, 'result': result.result})

In FastAPI, the same pattern works. Because the result backend is shared, you can query task status from either framework.

Error handling in tasks should be explicit. Use try/except inside the task and log the exception. Celery's task_acks_late and task_reject_on_worker_lost settings control how failures are handled. For example, setting task_acks_late = True ensures that a task is not acknowledged until it completes, so it can be retried if the worker crashes.

Production Considerations: Workers, Concurrency, and Monitoring

Running a Celery cluster in production requires attention to worker configuration. The --concurrency option controls how many tasks a worker processes simultaneously. For CPU-bound tasks, set concurrency to the number of CPU cores. For I/O-bound tasks, you can increase it, but be aware of memory usage.

celery -A myproject worker --loglevel=info --concurrency=4

If you use both Django and FastAPI in the same project, you can run a single worker pool for both. However, if task queues have different priorities, consider separate queues and workers. For example, you can route email tasks to a email queue and image processing to a images queue.

Monitoring is essential. Flower is a web-based tool that shows task progress, worker status, and queue lengths. Run it alongside your workers:

celery -A myproject flower --port=5555

This gives you visibility into task failures and bottlenecks. Also, configure logging in your tasks to capture exceptions. Celery's task_soft_time_limit and task_time_limit prevent tasks from hanging indefinitely.

Finally, ensure that your broker (Redis or RabbitMQ) is persistent and properly configured. If you use Redis, set a maxmemory policy that does not evict task messages. Use a separate database or a dedicated Redis instance for Celery to avoid interference with other application data.