FastAPI

FastAPI Long-Running Tasks — Real-Time Progress Tracking

Thirdy Gayares
14 min read

🎓 What You Will Learn

  • WebSockets: Real-time bidirectional communication
  • Progress Tracking: Update clients on task progress
  • Celery + Redis: Scalable long-running task execution
  • Retry Logic: Handle failures with exponential backoff
  • Timeout Handling: Gracefully manage task timeouts
  • Monitoring: Track task health and performance
FastAPIWebSocketsCelery

1Challenges of Long-Running Tasks

Long-running tasks create UX problems. Users do not know if anything is happening. HTTP connections time out. And there is no easy way to see progress.

Solution: Track progress in real-time and stream updates to clients using WebSockets.

2WebSocket Fundamentals

A WebSocket keeps one connection open. The server and the client can send messages at any time. This is perfect for sending progress updates. The endpoint below reads progress from TaskService, which we will build in the next section.

app/routes/tasks.py
import asyncio

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

from app.services.task_service import TaskService

app = FastAPI()


@app.websocket("/ws/task/{task_id}")
async def websocket_endpoint(websocket: WebSocket, task_id: str):
    await websocket.accept()

    try:
        while True:
            # Get task progress from Redis
            data = TaskService.get_progress(task_id)
            progress = data["progress"] if data else 0
            status = data["status"] if data else "pending"

            # Send to client
            await websocket.send_json({
                "task_id": task_id,
                "progress": progress,
                "status": status,
            })

            if status in ("completed", "failed"):
                break  # stop streaming once the task is done

            await asyncio.sleep(1)  # update every second
    except WebSocketDisconnect:
        pass  # client closed the connection

Tip: the Redis client here is synchronous. Calls are fast, so this is fine for a demo. For a fully async app, use redis.asyncio instead.

⏳ Real-Time Progress Bar Demo

Processing file...35%
✅ Real-time progress updates keep users engaged!

3Storing Progress in Redis

Use Redis to store task progress so multiple clients can track the same task.

app/services/task_service.py
import json
from datetime import UTC, datetime

import redis

redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)


class TaskService:
    @staticmethod
    def update_progress(task_id: str, progress: int, status: str = "running"):
        data = {
            "progress": progress,
            "status": status,
            "updated_at": datetime.now(UTC).isoformat(),
        }
        # Keys expire after 1 day so finished tasks do not pile up
        redis_client.set(f"task:{task_id}", json.dumps(data), ex=86400)

    @staticmethod
    def get_progress(task_id: str):
        data = redis_client.get(f"task:{task_id}")
        return json.loads(data) if data else None

4Celery Task Progress Reporting

Report progress from inside the Celery task. We use self.update_state() for Celery's own state, and we also write to Redis with TaskService so the WebSocket endpoint can read it.

app/celery_app.py
import csv

from celery import Celery

from app.services.task_service import TaskService

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


@celery_app.task(bind=True)
def process_large_file(self, file_path: str):
    with open(file_path, newline="") as f:
        rows = list(csv.reader(f))

    total_items = len(rows)

    for idx, row in enumerate(rows):
        handle_row(row)  # replace with your own processing logic

        # Update progress in Celery and in Redis
        progress = int((idx + 1) / total_items * 100)
        self.update_state(
            state="PROGRESS",
            meta={"current": idx + 1, "total": total_items, "progress": progress},
        )
        TaskService.update_progress(self.request.id, progress)

    TaskService.update_progress(self.request.id, 100, status="completed")
    return {"status": "Complete", "items_processed": total_items}

5Implementing Retry Logic

Retry failed tasks with exponential backoff. Backoff means each retry waits longer (1s, 2s, 4s, ...). This helps when a failure is temporary, like a slow network.

app/celery_app.py
@celery_app.task(
    bind=True,
    autoretry_for=(Exception,),
    retry_kwargs={"max_retries": 3},
    retry_backoff=True,       # wait 1s, 2s, 4s, ... between retries
    retry_backoff_max=600,    # never wait more than 10 minutes
    retry_jitter=True,        # add randomness so retries do not stack up
)
def resilient_task(self, data: dict):
    # Any exception raised here triggers an automatic retry
    return process_data(data)  # replace with your own logic

6Handling Task Timeouts

Set timeouts to prevent tasks from running indefinitely.

app/celery_app.py
from celery.exceptions import SoftTimeLimitExceeded


@celery_app.task(
    time_limit=3600,       # Hard limit: 1 hour, worker is killed
    soft_time_limit=3300,  # Soft limit: 55 minutes, exception is raised
)
def time_limited_task(data: dict):
    try:
        return long_running_operation(data)  # replace with your own logic
    except SoftTimeLimitExceeded:
        # You get 5 minutes to clean up before the hard limit
        save_partial_results()  # replace with your own cleanup logic
        raise

7Monitoring with Flower

Use Flower to monitor Celery tasks in real-time.

Setup: Install it with pip install flower, run celery -A app.celery_app flower, then visit http://localhost:5555

8HTTP Polling Alternative to WebSockets

For simpler cases, clients can poll for status via HTTP instead of WebSockets.

app/routes/tasks.py
from fastapi import APIRouter, HTTPException

from app.services.task_service import TaskService

router = APIRouter()


@router.get("/task/{task_id}/status")
def get_task_status(task_id: str):
    progress = TaskService.get_progress(task_id)
    if progress is None:
        raise HTTPException(status_code=404, detail="Task not found")

    return {
        "task_id": task_id,
        "progress": progress.get("progress", 0),
        "status": progress.get("status", "unknown"),
    }

9Job Queue Architecture

Implement a reliable job queue with retries and persistence.

app/models/job.py
from datetime import UTC, datetime

from sqlmodel import Field, SQLModel


def utc_now() -> datetime:
    return datetime.now(UTC)


class Job(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    task_id: str
    status: str = "pending"  # pending, running, completed, failed
    progress: int = 0
    error_message: str | None = None
    created_at: datetime = Field(default_factory=utc_now)
    updated_at: datetime = Field(default_factory=utc_now)
    retry_count: int = 0

10Dead Letter Queue for Failed Tasks

Move tasks to a dead letter queue after all retries are exhausted.

app/celery_app.py
import json

from app.services.task_service import redis_client


def send_to_dead_letter_queue(data: dict, error: str) -> None:
    redis_client.lpush("dead_letter_queue", json.dumps({"data": data, "error": error}))


@celery_app.task(bind=True, max_retries=3)
def critical_task(self, data: dict):
    try:
        return process(data)  # replace with your own logic
    except Exception as exc:
        if self.request.retries < self.max_retries:
            raise self.retry(exc=exc, countdown=60)

        # All retries used up: keep the payload for manual review
        send_to_dead_letter_queue(data, str(exc))
        raise

11Load Balancing Across Workers

Distribute tasks fairly across multiple workers using prefetch settings.

app/celery_app.py
# Each worker takes only 1 task at a time (fair for long tasks)
celery_app.conf.worker_prefetch_multiplier = 1

# Restart a worker process after 1000 tasks (avoids memory leaks)
celery_app.conf.worker_max_tasks_per_child = 1000

12Testing Long-Running Tasks

Test tasks synchronously during development and testing.

tests/test_tasks.py
def test_long_running_task():
    from app.celery_app import celery_app

    celery_app.conf.task_always_eager = True

    result = process_large_file.delay("test_file.csv")

    assert result.successful()
    assert result.result["items_processed"] > 0

13Common Pitfalls

  • No timeout: Tasks can hang indefinitely
  • No retry: Transient failures cause permanent failure
  • No monitoring: Can't debug stuck tasks
  • No idempotency: Retries cause side effects
  • Blocking WebSockets: Hold connections open unnecessarily

Performance: WebSocket connections are stateful and consume server resources. Use HTTP polling for non-critical updates to reduce load.

14Advanced Patterns

  • Task Groups: Run multiple tasks in parallel
  • Pipelines: Chain task results together
  • Chords: Run tasks then aggregate results
  • Custom Routing: Send tasks to specific workers

15Resources & What's Next

You now understand how to handle long-running tasks with real-time progress tracking. Use these patterns for file processing, data imports, report generation, and batch operations.

Next Topics: Database migrations, security authentication, and deployment strategies.

Congratulations! Your application can now handle complex, long-running operations while keeping users informed. Build with confidence! 🚀

About the Author

TG

Thirdy Gayares

Passionate developer creating custom solutions for everyone. I specialize in building user-friendly tools that solve real-world problems while maintaining the highest standards of security and privacy.

LinkedInWebsiteGitHubYouTube