FastAPI with Celery: Asynchronous Task Queue Implementation Guide

FastAPI is a modern, fast (high-performance) web framework for building APIs with Python 3.7+ based on standard Python type hints. Celery is an asynchronous task queue/job queue based on distributed message passing. Combining FastAPI with Celery allows developers to handle long-running tasks asynchronously, improving application responsiveness and scalability.

Why Use FastAPI with Celery?

FastAPI provides a quick and efficient way to develop APIs, but it operates synchronously by default. When dealing with resource-intensive or time-consuming tasks, such as sending emails, processing images, or performing database operations, using Celery enables these tasks to run in the background. This separation improves user experience and system performance.

Setting Up the Environment

To implement FastAPI with Celery, you need to install several packages:

fastapi
uvicorn
celery
redis

You can install them using pip:

pip install fastapi uvicorn celery redis

Configuring Celery with FastAPI

First, create a celery.py file to configure Celery:

from celery import Celery

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

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

Creating the FastAPI Application

Next, set up your FastAPI app to call Celery tasks asynchronously:

from fastapi import FastAPI
from celery import Celery
from celery.result import AsyncResult

app = FastAPI()

celery_app = Celery('worker', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')

@app.post('/add/')
async def perform_add(x: int, y: int):
    task = celery_app.send_task('add', args=[x, y])
    return {'task_id': task.id}

@app.get('/result/{task_id}')
async def get_result(task_id: str):
    result = AsyncResult(task_id, app=celery_app)
    if result.ready():
        return {'status': 'completed', 'result': result.get()}
    else:
        return {'status': 'pending'}

Running the System

Start the Redis server:

redis-server

Run the Celery worker:

celery -A celery worker --loglevel=info

Finally, run the FastAPI server:

uvicorn main:app --reload

Testing the Implementation

Use a tool like cURL, Postman, or your browser to interact with the API:

Send a POST request to /add/ with JSON body:

POST /add/
Content-Type: application/json

{
  "x": 5,
  "y": 10
}

You will receive a task ID, which you can use to check the result:

Send a GET request to /result/{task_id} to see if the task is complete and get the result.

Best Practices and Tips

Configure Celery to use a reliable message broker like Redis or RabbitMQ.
Handle task failures gracefully with retries and error logging.
Use task result backends to track progress and results.
Secure your message broker and API endpoints.

Integrating FastAPI with Celery creates a powerful architecture for handling asynchronous tasks, improving performance, and providing a better user experience.

Table of Contents