Event-driven architecture and message queues

Difficulty: Advanced

Question

How do you integrate FastAPI with message queues for event-driven processing?

Answer

For heavy or long-running tasks that exceed what `BackgroundTasks` can handle, integrate with a message queue:

Celery + Redis/RabbitMQ - The most popular Python task queue. Define tasks as decorated functions, call them asynchronously from FastAPI routes. Celery workers run separately and process tasks from the queue.

ARQ (Async Redis Queue) - Lightweight async alternative to Celery. Works naturally with FastAPI's async architecture.

Kafka - For high-throughput event streaming. FastAPI produces messages; separate consumers process them. Used for analytics, audit logs, cross-service communication.

Pattern: FastAPI handles the HTTP request, publishes an event/task to the queue, and returns immediately. The worker processes the task asynchronously. Use webhooks or polling for the client to check task status.

Code examples

Celery integration with FastAPI

# tasks.py
from celery import Celery

celery_app = Celery("tasks", broker="redis://localhost:6379/0", backend="redis://localhost:6379/1")

@celery_app.task
def process_video(video_id: int):
    # Heavy processing - runs in Celery worker, not FastAPI
    import time
    time.sleep(30)  # simulate long processing
    return {"video_id": video_id, "status": "processed"}

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

app = FastAPI()

@app.post("/videos/{video_id}/process")
def start_processing(video_id: int):
    task = process_video.delay(video_id)  # enqueue task
    return {"task_id": task.id, "status": "queued"}

@app.get("/tasks/{task_id}")
def get_task_status(task_id: str):
    result = process_video.AsyncResult(task_id)
    return {
        "task_id": task_id,
        "status": result.status,
        "result": result.result if result.ready() else None
    }

Key points

• Run Celery workers separately: `celery -A tasks worker --loglevel=info`
• Use Celery Beat for scheduled/periodic tasks (cron-like)
• For microservices communication, consider event-driven patterns over direct HTTP calls

Concepts covered

message queue, Celery, RabbitMQ, Kafka, event-driven