How to Handle Long Running Tasks in FastAPI with Callbacks

In this blog, we will explore handling long-running tasks in FastAPI while allowing users to receive notifications once the task is complete. For this purpose, we will use FastAPI’s background tasks feature and implement a callback mechanism.

FastAPI Background Tasks

FastAPI has built-in support for background tasks using the BackgroundTasks class. This feature allows long-running tasks to execute in the background without blocking the main application, enabling it to continue processing other incoming requests.

To use background tasks in FastAPI, follow these steps:

1. Import BackgroundTasks from FastAPI.
2. Create a new function (e.g., long_running_task) that performs the time-consuming operation.
3. Modify your route to accept a BackgroundTasks parameter.
4. In your route, add the task using background_tasks.add_task() method with the required arguments, and immediately return a response to the user.

Setting up Callbacks

To notify the user once the task is complete, we can implement a callback mechanism by allowing the user to provide a callback URL as a parameter in their request. When the long-running task finishes, our application will send an HTTP request to the specified callback URL with the task’s results or status update.

In this tutorial, we will use Python’s httpx library to make HTTP requests.

Complete Example

Here’s the complete code example, including the FastAPI application with background tasks and callbacks and a client-side script to call the API endpoint.

Prerequisites

Install the necessary packages:

pip install fastapi uvicorn httpx requests

FastAPI Application

Create a new FastAPI application in a file named app.py:

from fastapi import FastAPI, BackgroundTasks
from pydantic import HttpUrl
import time
import httpx

app = FastAPI()

async def post_callback(url: str, payload: dict):
    async with httpx.AsyncClient() as client:
        await client.post(url, json=payload)

async def long_running_task(duration: int, callback_url: str):
    # Simulate a time-consuming operation
    time.sleep(duration)

    # Send the result to the callback URL
    result = {"status": "completed", "data": "result_data"}
    await post_callback(callback_url, result)

@app.post("/start_task")
async def start_long_running_task(background_tasks: BackgroundTasks, duration: int = 10, callback_url: str = "http://example.com/callback"):
    background_tasks.add_task(long_running_task, duration, callback_url)
    return {"message": "Task started! You'll receive a notification at the callback URL once it's complete."}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("app:app", host="127.0.0.1", port=8000, log_level="info", reload=True, workers=1)

Calling the API Endpoint

Create a new Python script named call_start_task.py to call the /start_task endpoint:

import requests

url = "http://127.0.0.1:8000/start_task"
data = {
    "duration": 10,
    "callback_url": "http://example.com/callback"
}

response = requests.post(url, json=data)

print(response.status_code)
print(response.json())

Replace the http://example.com/callback with your own callback URL that accepts POST requests and processes the data sent as JSON payload.

Testing the Example

First, start your FastAPI application by running:

python app.py

Next, test the API by running the client-side script:

python call_start_task.py

This script will call the /start_task endpoint, providing a duration and callback URL. The FastAPI app will then run the long-running task in the background and send a POST request with the results to the specified callback URL upon task completion.

This approach allows your FastAPI application to effectively handle long-running tasks while maintaining excellent performance and providing a way to notify users once their tasks are complete.