Background Tasks & Workers

FastAPI BackgroundTasks

For quick fire-and-forget work (< 30 seconds, no retry needed):

from fastapi import BackgroundTasks


@router.post("/users/{user_id}/send-report")
async def send_report(
    user_id: UUID,
    background_tasks: BackgroundTasks,
    service: ReportService = Depends(get_report_service),
):
    report = await service.generate_report(user_id)
    background_tasks.add_task(service.send_report_email, user_id, report)
    return {"status": "Report is being sent"}

When to Use What

Need	Use
Quick side effect, no retry	BackgroundTasks
Long-running work (> 30s)	Task queue (Celery, ARQ)
Needs retry on failure	Task queue
Needs progress tracking	Task queue
Scheduled/periodic jobs	Celery Beat or cron

Task Queue Pattern

from celery import Celery

celery_app = Celery("worker", broker=settings.redis_url)


@celery_app.task(bind=True, max_retries=3, default_retry_delay=60)
def process_video(self, video_id: str):
    try:
        # Long-running processing
        ...
    except TransientError as exc:
        raise self.retry(exc=exc)

---

See Also

• Logging -- Structured logging for tracking background task execution
• Error Handling -- Exception hierarchy and retry patterns for task failures