FastAPI Background Task Processing

This skill provides patterns for background task processing with multiple frameworks: ARQ (recommended for async), Celery, and Dramatiq.

ARQ (Async Redis Queue) - Recommended

Installation

pip install arq

Configuration

app/workers/config.py

from arq.connections import RedisSettings from app.config import get_settings

settings = get_settings()

class WorkerSettings: redis_settings = RedisSettings( host=settings.redis_host, port=settings.redis_port, password=settings.redis_password, database=1 # Separate from cache )

# Job settings
max_jobs = 10
job_timeout = 300  # 5 minutes
keep_result = 3600  # 1 hour
queue_name = "default"

# Cron jobs
cron_jobs = []

Task Definitions

app/workers/tasks.py

from arq import cron from typing import Dict, Any import asyncio

async def send_email(ctx: Dict[str, Any], to: str, subject: str, body: str): """Send email asynchronously.""" email_service = ctx.get("email_service") await email_service.send(to=to, subject=subject, body=body) return {"status": "sent", "to": to}

async def process_upload(ctx: Dict[str, Any], file_id: str, user_id: str): """Process uploaded file (resize, convert, etc.).""" storage = ctx.get("storage") file_data = await storage.get(file_id)

# Process file
processed = await process_file(file_data)

# Save processed file
await storage.put(f"processed/{file_id}", processed)

return {"status": "processed", "file_id": file_id}

async def cleanup_expired(ctx: Dict[str, Any]): """Periodic cleanup of expired data.""" db = ctx.get("db") result = await db.delete_expired() return {"deleted": result.deleted_count}

Cron job example

@cron(hour=2, minute=0) # Run at 2 AM daily async def daily_report(ctx: Dict[str, Any]): """Generate daily report.""" report_service = ctx.get("report_service") await report_service.generate_daily()

Worker Entry Point

app/workers/main.py

from arq import create_pool from arq.connections import RedisSettings from app.workers.config import WorkerSettings from app.workers.tasks import send_email, process_upload, cleanup_expired, daily_report from app.infrastructure.database import init_database from app.services.email import EmailService

async def startup(ctx: Dict[str, Any]): """Worker startup - initialize services.""" await init_database() ctx["email_service"] = EmailService() ctx["db"] = get_db()

async def shutdown(ctx: Dict[str, Any]): """Worker shutdown - cleanup.""" await close_database()

class WorkerSettings(WorkerSettings): functions = [send_email, process_upload, cleanup_expired] cron_jobs = [daily_report] on_startup = startup on_shutdown = shutdown

Run with: arq app.workers.main.WorkerSettings

Enqueueing Tasks from FastAPI

app/dependencies.py

from arq import ArqRedis, create_pool from arq.connections import RedisSettings

async def get_task_queue() -> ArqRedis: return await create_pool(RedisSettings())

app/routes/users.py

from fastapi import Depends from arq import ArqRedis

@router.post("/users/{user_id}/welcome") async def send_welcome_email( user_id: str, queue: ArqRedis = Depends(get_task_queue) ): user = await get_user(user_id)

# Enqueue background task
job = await queue.enqueue_job(
    "send_email",
    to=user.email,
    subject="Welcome!",
    body="Thanks for signing up."
)

return {"job_id": job.job_id, "status": "queued"}

@router.post("/uploads") async def upload_file( file: UploadFile, user: User = Depends(get_current_user), queue: ArqRedis = Depends(get_task_queue) ): # Save file file_id = await save_file(file)

# Enqueue processing
await queue.enqueue_job(
    "process_upload",
    file_id=file_id,
    user_id=str(user.id),
    _defer_by=5  # Delay 5 seconds
)

return {"file_id": file_id, "status": "processing"}

Celery (Battle-Tested)

Configuration

app/workers/celery_app.py

from celery import Celery from app.config import get_settings

settings = get_settings()

celery_app = Celery( "worker", broker=settings.celery_broker_url, backend=settings.celery_result_backend, include=["app.workers.celery_tasks"] )

celery_app.conf.update( task_serializer="json", accept_content=["json"], result_serializer="json", timezone="UTC", enable_utc=True, task_track_started=True, task_time_limit=300, worker_prefetch_multiplier=1, )

Periodic tasks (Celery Beat)

celery_app.conf.beat_schedule = { "cleanup-every-hour": { "task": "app.workers.celery_tasks.cleanup_expired", "schedule": 3600.0, }, "daily-report": { "task": "app.workers.celery_tasks.generate_daily_report", "schedule": crontab(hour=2, minute=0), }, }

Celery Tasks

app/workers/celery_tasks.py

from app.workers.celery_app import celery_app import asyncio

def run_async(coro): """Helper to run async code in sync Celery tasks.""" loop = asyncio.get_event_loop() return loop.run_until_complete(coro)

@celery_app.task(bind=True, max_retries=3) def send_email(self, to: str, subject: str, body: str): try: run_async(_send_email_async(to, subject, body)) return {"status": "sent", "to": to} except Exception as exc: self.retry(exc=exc, countdown=60)

@celery_app.task def process_upload(file_id: str, user_id: str): run_async(_process_upload_async(file_id, user_id)) return {"status": "processed", "file_id": file_id}

Dramatiq (Modern Celery Alternative)

Configuration

app/workers/dramatiq_app.py

import dramatiq from dramatiq.brokers.redis import RedisBroker from dramatiq.results import Results from dramatiq.results.backends import RedisBackend

redis_broker = RedisBroker(url="redis://localhost:6379/0") result_backend = RedisBackend(url="redis://localhost:6379/1")

redis_broker.add_middleware(Results(backend=result_backend)) dramatiq.set_broker(redis_broker)

Dramatiq Tasks

app/workers/dramatiq_tasks.py

import dramatiq

@dramatiq.actor(max_retries=3, min_backoff=1000) def send_email(to: str, subject: str, body: str): # Sync implementation return {"status": "sent", "to": to}

@dramatiq.actor(time_limit=300000) # 5 min timeout def process_upload(file_id: str, user_id: str): return {"status": "processed", "file_id": file_id}

FastAPI Built-in Background Tasks

For simple fire-and-forget tasks (no persistence):

from fastapi import BackgroundTasks

async def write_log(message: str): with open("log.txt", "a") as f: f.write(f"{message}\n")

@router.post("/log") async def create_log(message: str, background_tasks: BackgroundTasks): background_tasks.add_task(write_log, message) return {"status": "logged"}

Additional Resources

Reference Files

For detailed patterns:

• references/arq-advanced.md

• ARQ advanced patterns, retries, priorities

• references/celery-patterns.md

• Celery best practices, chains, groups

• references/monitoring.md

• Flower, task monitoring

Example Files

Working examples in examples/ :

• examples/arq_worker.py

• Complete ARQ worker

• examples/celery_app.py

• Celery configuration

• examples/task_service.py

• Task enqueueing service