Handling Long-Running Tasks in FastAPI, Best Practices and Strategies

Building fast and efficient APIs is an essential aspect of modern software development. However, it’s not uncommon to encounter API endpoints that take a long time to execute due to reasons such as data processing, calls to third-party services, or complex computations. When such scenarios arise, it’s crucial to ensure that these long-running tasks don’t degrade user experience or system performance. This blog post aims to guide you through the best practices and strategies for managing long-running tasks in FastAPI.

The Challenges of Long-Running API Endpoints

1. User Experience: A prolonged wait time for a response can result in a poor user experience.
2. Resource Utilization: Long-running tasks can consume significant system resources, potentially affecting the performance of other tasks.
3. Error Handling: Tasks that take a long time to complete are more susceptible to errors, requiring robust error-handling mechanisms.

Best Practices for Managing Long-Running Tasks

1. Asynchronous Endpoints

You can use Python’s async def syntax to define asynchronous endpoints in FastAPI, which can help in I/O-bound operations.

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    # Perform a long-running task
    return {"message": "done"}

2. Background Tasks

FastAPI allows you to run background tasks that can continue processing after the response has been sent.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def long_running_task():
    # Perform a long-running task
    pass

@app.get("/")
async def read_root(background_tasks: BackgroundTasks):
    background_tasks.add_task(long_running_task)
    return {"message": "Task is running in the background"}

3. Using Celery for Distributed Task Queues

For particularly long-running tasks, you can offload them to a task queue like Celery.

from fastapi import FastAPI
from celery import Celery

app = FastAPI()
celery_app = Celery('tasks', broker='pyamqp://guest@localhost//')

@celery_app.task
def long_running_task():
    # Perform a long-running task
    pass

@app.get("/")
def read_root():
    long_running_task.apply_async()
    return {"message": "Task is running in the background"}

Strategies to Notify Users

Once a task is offloaded or made asynchronous, it’s essential to inform the user of its completion. Below are some strategies to achieve this:

1. Polling

In this approach, the client initially receives a task ID and then repeatedly polls an endpoint to check the task’s status.

Server-side:

from fastapi import FastAPI
from some_task_queue import some_task_queue

app = FastAPI()

@app.post("/start_task/")
def start_task():
    task_id = some_task_queue.enqueue("long_running_task")
    return {"task_id": task_id}

@app.get("/get_result/{task_id}")
def get_result(task_id: str):
    result = some_task_queue.get_result(task_id)
    return {"result": result}

Client-side:

async function startAndPollTask() {
  const response = await fetch('/start_task/');
  const task = await response.json();
  
  let result;
  do {
    const resultResponse = await fetch(`/get_result/${task.task_id}`);
    result = await resultResponse.json();
    if (result.is_done) {
      break;
    }
    await new Promise(resolve => setTimeout(resolve, 2000));
  } while(true);
  
  console.log("Final result:", result);
}

2. Webhooks

Here, the client provides a callback URL that the server can POST the result to once the task is complete.

Server-side:

from fastapi import FastAPI, BackgroundTasks
import requests

app = FastAPI()

def long_running_task(callback_url):
    # Your long-running task here
    result = "some_result"
    requests.post(callback_url, json={"result": result})

@app.post("/start_task/")
async def start_task(background_tasks: BackgroundTasks, callback_url: str):
    background_tasks.add_task(long_running_task, callback_url)
    return {"status": "Task started"}

3. WebSockets

You can establish a WebSocket connection between the client and server to send the result when the task is complete.

Server-side:

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws/")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    # Long-running task here
    result = "some_result"
    await websocket.send_json({"result": result})

Client-side:

const socket = new WebSocket('ws://localhost:8000/ws/');

socket.addEventListener('message', function(event) {
    const result = JSON.parse(event.data);
    console.log("Received result:", result);
});

4. Server-Sent Events (SSE)

SSE allows the server to send updates and final results over a single HTTP connection.

Server-side:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.get("/task_status/")
def get_status():
    def event_stream():
        # Long-running task
        result = "some_result"
        yield f"data: {result}\n\n"
    return StreamingResponse(event_stream(), media_type="text/event-stream")

Client-side:

const eventSource = new EventSource('/task_status/');

eventSource.onmessage = function(event) {
    const result = event.data;
    console.log("Received result:", result);
};

Conclusion

Long-running tasks can pose challenges in API design, but FastAPI provides a variety of features and techniques to handle them efficiently. Whether it’s asynchronous programming, background tasks, or advanced strategies like Celery, Webhooks, and WebSockets, you can choose the right approach based on your API’s requirements. By adhering to these best practices and strategies, you can ensure that long-running tasks are managed effectively without compromising user experience or system performance.