Leveraging Background Tasks in FastAPI: Enhancing Performance and Responsiveness

TLDR;

Utilizing FastAPI’s BackgroundTasks feature allows for the efficient management of long-running processes. By moving these tasks to the background, applications can maintain responsiveness and provide a seamless user experience, all without compromising performance or scalability.

Introduction

FastAPI, a leading-edge web framework for crafting APIs in Python, boasts superior speed, user-friendly attributes, and outstanding asynchronous capabilities. Among these, a standout feature is FastAPI’s BackgroundTasks — an innovative tool engineered for managing long-running, time-consuming tasks without inhibiting the primary application operations.

In this blog post, we will dig deep into FastAPI’s BackgroundTasks, shedding light on how they can significantly improve application performance and responsiveness. We’ll chronicle this journey through a step-by-step enhancement of a simple FastAPI application, demonstrating the transformation when long-running tasks are meticulously offloaded to the background.

Creating Simple FastAPI application

In our given Python code snippet, we showcase a basic FastAPI application with a key challenge — handling a long-running operation, simulated by very_long_task(). This function introduces a delay, demonstrating how a time-consuming task can block an application.

import time
import uvicorn
from fastapi import FastAPI

app = FastAPI()

def very_long_task():
   print("Starting very long task")
   time.sleep(5)
   print("Ending very long task")

@app.post("/long_task")
def long_task_endpoint():
   print("Entering endpoint ")
   very_long_task()
   print("Exiting endpoint ")
   return {"status": "very_long_task ended"}

if __name__ == "__main__":
   uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True)

When calling the long_task endpoint, you'll observe the following log sequence:

Entering endpoint
Starting very long task
Ending very long task
Exiting endpoint

The snag in this implementation is that the connection to our application hangs during the execution of very_long_task. Keeping the connection open this long is typically considered bad practice. If the connection breaks during this time, the long task won't be executed, resulting in potential data inconsistency or other undesired effects.

We’ll now explore how FastAPI’s BackgroundTasks can be used in handling this scenario. Stay tuned as we dive deeper into the world of asynchronous programming in FastAPI, improving our application’s responsiveness and reliability.

Solving the blocking problem

The use of BackgroundTasks in FastAPI significantly alters how our application handles time-consuming operations, effectively improving performance and responsiveness. This modification has turned our blocking operation into a non-blocking one, allowing the application to continue processing other requests while the lengthy operation runs in the background.

import time
from collections import OrderedDict

import uvicorn
from fastapi import BackgroundTasks, FastAPI, HTTPException

app = FastAPI()

def very_long_task():
   print("Starting very long task")
   time.sleep(5)
   print("Ending very long task")

@app.post("/long_task")
def long_task_endpoint(background_tasks: BackgroundTasks):
   print("Entering endpoint ")
   background_tasks.add_task(very_long_task)
   print("Exiting endpoint ")
   return {"status": "very long task is running"}

if __name__ == "__main__":
   uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True)

Here’s a deeper look at the modifications made:

Our long_task_endpoint() function now accepts an additional parameter, background_tasks, of type BackgroundTasks. This parameter allows FastAPI to schedule functions to be executed in the background after returning a response.

Inside long_task_endpoint(), we use the add_task() method on the background_tasks object to add our very_long_task() function to the list of background tasks. Instead of executing very_long_task() directly and waiting for it to finish, our function now simply adds it to the background tasks and moves on, allowing it to return a response immediately.

We updated the response as well: instead of informing the client that the task has ended, we now return a response indicating that the long task is running. This is because the task is now scheduled to run in the background, and the response is sent before the task completes.

In essence, this updated FastAPI application shows the powerful potential of background tasks for managing long-running operations. With this approach, our application becomes significantly more efficient and responsive, capable of handling multiple requests without being hindered by time-consuming tasks.

Bonus: Checking out the status of the task

In the latest iteration, we have introduced a new GET method. The endpoint /status now enables users to check the status of a long-running task, providing insights into the progress of specific operations within the application.

import time
import uuid
import uvicorn
from fastapi import BackgroundTasks, FastAPI, HTTPException

app = FastAPI()

status_dict = {}

def very_long_task(task_id):
   print("Starting very long task")
   try:
       time.sleep(5)
       status_dict[str(task_id)] = "completed"
   except Exception as e:
       status_dict[str(task_id)] = "failed"
   print("Ending very long task")

@app.get("/status")
def get_status(task_id: str):
   if task_id not in status_dict:
       raise HTTPException(status_code=404, detail="Item not found")
   return status_dict[task_id]

@app.post("/long_task")
def long_task_endpoint(background_tasks: BackgroundTasks):
   print("Entering endpoint ")
   task_id = uuid.uuid1()
   status_dict[str(task_id)] = "running"
   background_tasks.add_task(very_long_task, task_id)
   print("Exiting endpoint ")
   return task_id

if __name__ == "__main__":
   uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True)

Major updates in this iteration include the following enhancements:

• We introduced an endpoint, /status, which accepts the task_id that we want to use to check the status of a desired task. If the task is not found, an HTTPException is raised, resulting in a 404 error.
• To enable this functionality, we’ve created a status_dict dictionary that keeps track of the statuses for all tasks.
• Within the long_task endpoint, a task_id is generated, then added to the status_dict with an initial status of 'running.' This task_id is also passed to the main task function very_long_task, where the main operation is executed.
• Upon successful execution of the task, the status in status_dict is updated to 'completed.' If a failure occurs during execution, the status is set to 'failed.'

Remember that in real-life use cases, you shouldn’t store the statuses in a global variable. The problem with this approach is that in case of restarting the application, you would lose all the statuses. Additionally, you don’t have control over the size of status_dict; in critical cases, it could overflow the memory on the host. A better approach would be to use a database to store the statuses. Another option is to leverage queue systems, such as Celery, to manage these tasks more efficiently.

Conclusion

In this article, we explored a more advanced topic in FastAPI implementation: background tasks. This feature enabled us to optimize the performance of our application by transforming a long, blocking operation into a non-blocking one. Additionally, we implemented a system to check the status of ongoing tasks, ensuring a more responsive and user-friendly application. These techniques represent a powerful way to enhance both the efficiency and functionality of FastAPI-based projects.