Background Tasks and Celery Integration in FastAPI

Why Background Tasks?

Some operations should not block an HTTP response. Sending emails, generating reports, processing uploads, syncing with third-party APIs — these can take seconds or minutes. Background tasks let you return a response immediately and handle the heavy work asynchronously.

FastAPI gives you two paths: the built-in BackgroundTasks for simple in-process work, and Celery for distributed, reliable job processing.

FastAPI Built-in BackgroundTasks

For lightweight tasks that run in the same process, FastAPI provides BackgroundTasks. It runs functions after the response is sent.

from fastapi import BackgroundTasks, FastAPI
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

def send_welcome_email(email: str, username: str):
    # Simulate sending an email
    logger.info(f"Sending welcome email to {email}")
    # In production, use an email service like SendGrid or SES
    import time
    time.sleep(2)  # Simulates API call
    logger.info(f"Welcome email sent to {username} at {email}")

def write_audit_log(user_id: int, action: str):
    logger.info(f"Audit: user {user_id} performed {action}")
    # Write to audit log database table

@app.post("/register")
async def register_user(
    username: str,
    email: str,
    background_tasks: BackgroundTasks
):
    # Create user in database (fast)
    user_id = await create_user(username, email)

    # Queue background work (does not block response)
    background_tasks.add_task(send_welcome_email, email, username)
    background_tasks.add_task(write_audit_log, user_id, "registered")

    return {"user_id": user_id, "message": "Registration successful"}

The response returns immediately. The email and audit log tasks run after the response is sent, in the same process. You can add multiple tasks and they execute sequentially.

BackgroundTasks with Dependencies

Background tasks work with FastAPI’s dependency injection system.

from fastapi import Depends

async def get_email_service():
    service = EmailService()
    yield service
    await service.close()

@app.post("/invite")
async def invite_user(
    email: str,
    background_tasks: BackgroundTasks,
    email_service=Depends(get_email_service)
):
    background_tasks.add_task(
        email_service.send_invitation,
        recipient=email
    )
    return {"message": f"Invitation queued for {email}"}

Limitations of BackgroundTasks

Built-in background tasks have important constraints. They run in the same process, so a worker crash loses all pending tasks. There is no retry mechanism. There is no way to check task status. They cannot distribute work across multiple machines. For anything beyond fire-and-forget notifications, you need Celery.

Setting Up Celery with FastAPI

Celery is a distributed task queue that runs tasks in separate worker processes. It uses a message broker (Redis or RabbitMQ) to pass tasks between your FastAPI app and the workers.

Project Structure

project/
    app/
        __init__.py
        main.py          # FastAPI app
        tasks.py          # Celery tasks
        celery_app.py     # Celery configuration
        config.py         # Settings

Celery Configuration

# app/celery_app.py
from celery import Celery

celery = Celery(
    "worker",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

celery.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_track_started=True,
    task_acks_late=True,           # Tasks acknowledged after completion
    worker_prefetch_multiplier=1,  # One task at a time per worker
    result_expires=3600,           # Results expire after 1 hour
)

# Auto-discover tasks in the app package
celery.autodiscover_tasks(["app"])

Defining Celery Tasks

# app/tasks.py
from app.celery_app import celery
import time
import logging

logger = logging.getLogger(__name__)

@celery.task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,
    name="app.tasks.send_email"
)
def send_email(self, to: str, subject: str, body: str):
    try:
        logger.info(f"Sending email to {to}: {subject}")
        # Call email service API
        response = email_client.send(to=to, subject=subject, body=body)
        return {"status": "sent", "message_id": response.id}
    except ConnectionError as exc:
        logger.warning(f"Email send failed, retrying: {exc}")
        raise self.retry(exc=exc)

@celery.task(
    bind=True,
    name="app.tasks.generate_report"
)
def generate_report(self, report_type: str, user_id: int):
    self.update_state(state="PROCESSING", meta={"progress": 0})

    # Step 1: Fetch data
    data = fetch_report_data(report_type, user_id)
    self.update_state(state="PROCESSING", meta={"progress": 33})

    # Step 2: Generate file
    file_path = render_report(data, report_type)
    self.update_state(state="PROCESSING", meta={"progress": 66})

    # Step 3: Upload to storage
    url = upload_to_s3(file_path)
    self.update_state(state="PROCESSING", meta={"progress": 100})

    return {"url": url, "report_type": report_type}

The bind=True parameter gives the task access to self, which lets you update state and trigger retries. The max_retries and default_retry_delay configure automatic retry behavior.

Calling Celery Tasks from FastAPI

# app/main.py
from fastapi import FastAPI, HTTPException
from app.tasks import send_email, generate_report
from celery.result import AsyncResult

app = FastAPI()

@app.post("/send-notification")
async def send_notification(to: str, subject: str, body: str):
    task = send_email.delay(to=to, subject=subject, body=body)
    return {"task_id": task.id, "status": "queued"}

@app.post("/reports")
async def request_report(report_type: str, user_id: int):
    task = generate_report.delay(
        report_type=report_type,
        user_id=user_id
    )
    return {"task_id": task.id, "status": "queued"}

@app.get("/tasks/{task_id}")
async def get_task_status(task_id: str):
    result = AsyncResult(task_id)

    if result.state == "PENDING":
        return {"task_id": task_id, "status": "pending"}
    elif result.state == "PROCESSING":
        return {
            "task_id": task_id,
            "status": "processing",
            "progress": result.info.get("progress", 0)
        }
    elif result.state == "SUCCESS":
        return {
            "task_id": task_id,
            "status": "completed",
            "result": result.result
        }
    elif result.state == "FAILURE":
        return {
            "task_id": task_id,
            "status": "failed",
            "error": str(result.result)
        }
    return {"task_id": task_id, "status": result.state}

The .delay() method sends the task to the broker and returns immediately with a task ID. Clients can poll the status endpoint to track progress.

Running the Workers

Start Celery workers alongside your FastAPI server.

# Terminal 1: Start FastAPI
uvicorn app.main:app --reload

# Terminal 2: Start Celery worker
celery -A app.celery_app worker --loglevel=info --concurrency=4

# Terminal 3 (optional): Start Celery Beat for scheduled tasks
celery -A app.celery_app beat --loglevel=info

Scheduled Tasks with Celery Beat

Celery Beat handles periodic tasks — cron-like schedules for recurring jobs.

# app/celery_app.py (add to existing config)
from celery.schedules import crontab

celery.conf.beat_schedule = {
    "cleanup-expired-sessions": {
        "task": "app.tasks.cleanup_sessions",
        "schedule": crontab(minute=0, hour="*/2"),  # Every 2 hours
    },
    "daily-digest-email": {
        "task": "app.tasks.send_daily_digest",
        "schedule": crontab(minute=0, hour=8),  # Daily at 8 AM
    },
    "health-check-ping": {
        "task": "app.tasks.health_ping",
        "schedule": 30.0,  # Every 30 seconds
    },
}

When to Use Which

Use BackgroundTasks when the task is fast (under a few seconds), failure is acceptable, you do not need retries or status tracking, and the task only needs to run on the same machine.

Use Celery when tasks are long-running, you need retries with exponential backoff, you need to track task progress, you need to distribute work across multiple workers, or you need scheduled/periodic tasks.

A common pattern is to start with BackgroundTasks during development and migrate to Celery when you need reliability and scale.

Key Takeaways

FastAPI’s built-in BackgroundTasks handles simple fire-and-forget work with zero setup. Celery adds reliability, retries, progress tracking, and distributed execution at the cost of infrastructure complexity (a broker, workers, and optionally beat). Most production applications end up using both — BackgroundTasks for quick in-process work like logging, and Celery for everything that needs guarantees.