How to Create an Asynchronous API with FastAPI and Job Concept

Introduction

In this post, we will learn how to create an asynchronous API with FastAPI, a modern, fast, and high-performance web framework for building APIs with Python. We will use the concept of job to handle long-running or CPU-intensive tasks in the background, without blocking the main thread or the request-response cycle.

Why Asynchronous APIs?

Asynchronous APIs are APIs that do not provide data immediately, but rather at a later time. This can be useful for situations where the data is not available instantly, or where the processing of the data takes a long time. For example, if you want to send an email, generate a report, or perform some complex calculations, you might not want to wait for the result before moving on to the next task. Instead, you can use an asynchronous API to start the task in the background, and then check the status or the result later.

Asynchronous APIs can also improve the performance and scalability of your application, as they can handle multiple requests concurrently, and manage the communication intelligently between services. For example, if you have a microservice architecture, you can use asynchronous APIs to send and receive messages between different services, using a publish-subscribe pattern. This way, the services can subscribe to the events that matter to them, and receive updates without polling for new information.

How to Work

To create our asynchronous API with FastAPI and job concept, we need to follow these steps:

Install FastAPI and its dependencies
Create a FastAPI app
Define a global dictionary to store the results of the background tasks
Define a function to simulate a long-running task
Define an async endpoint to start a background task and return a job ID
Define another async endpoint to get the result of a background task using the job ID

Let’s see how to do each step in detail.

Install FastAPI and its dependencies

To install FastAPI and its dependencies, we can use the pip command in our terminal:

pip install fastapi[all]

This will install FastAPI along with other packages that we need, such as uvicorn, a fast ASGI server that can run our app.

Create a FastAPI app

To create a FastAPI app, we need to import FastAPI and BackgroundTasks from the fastapi module, and then create an instance of the FastAPI class:

# Import FastAPI and BackgroundTasks
from fastapi import FastAPI, BackgroundTasks

# Create a FastAPI app
app = FastAPI()

Define a global dictionary to store the results of the background tasks

To store the results of the background tasks, we can use a global dictionary that maps the job ID to the result. We can initialize the dictionary as an empty one:

# Define a global dictionary to store the results of the background tasks
results = {}

Define a function to simulate a long-running task

To simulate a long-running task, we can define a function that takes a job ID as a parameter, and then sleeps for 10 seconds before storing the result in the global dictionary. We can use the time module to implement the sleep function:

# Define a function to simulate a long-running task
def long_task(job_id: int):
    # Import time
    import time
    # Sleep for 10 seconds
    time.sleep(10)
    # Store the result in the global dictionary
    results[job_id] = f"Task {job_id} completed"

Define an async endpoint to start a background task and return a job ID

To define an async endpoint to start a background task and return a job ID, we can use the app.post decorator to specify the path and the method of the endpoint. We can also use the BackgroundTasks dependency to add the long-running function to the background tasks. We can generate a random job ID using the randint function from the random module, and then return a message to the user with the job ID:

# Define an async endpoint to start a background task and return a job ID
@app.post("/start")
async def start(background_tasks: BackgroundTasks):
    # Import random
    import random
    # Generate a random job ID
    job_id = random.randint(1, 100)
    # Add the long_task function to the background tasks
    background_tasks.add_task(long_task, job_id)
    # Return a message to the user with the job ID
    return {"message": f"Task {job_id} started"}

Define another async endpoint to get the result of a background task using the job ID

To define another async endpoint to get the result of a background task using the job ID, we can use the app.get decorator to specify the path and the method of the endpoint. We can also use a path parameter to get the job ID from the user. We can check if the job ID is in the global dictionary, and if yes, return the result. If not, we can return a message that the task is not found or not finished:

# Define another async endpoint to get the result of a background task using the job ID
@app.get("/get/{job_id}")
async def get(job_id: int):
    # Check if the job ID is in the global dictionary
    if job_id in results:
        # Return the result
        return {"result": results[job_id]}
    else:
        # Return a message that the task is not found or not finished
        return {"message": f"Task {job_id} not found or not finished"}

Demo

To run our asynchronous API with FastAPI and job concept, we can save our code in a file, such as main.py, and then run it using uvicorn in our terminal:

uvicorn main:app --reload

This will start our app on the localhost port 8000, and reload it automatically if we make any changes to the code.

We can test our API using a tool like curl or a browser. For example, we can start a task by sending a POST request to /start:

curl -X POST http://localhost:8000/start

We should get a response like this:

{"message": "Task 42 started"}

Then, we can get the result of the task by sending a GET request to /get/42:

curl http://localhost:8000/get/42

We should get a response like this:

{"result": "Task 42 completed"}

If we try to get the result of a task that is not started or not finished, we should get a message like this:

{"message": "Task 99 not found or not finished"}

Conclusion

In this post, we learned how to create an asynchronous API with FastAPI and job concept. We saw how we can use the BackgroundTasks dependency to add functions to the background tasks, and how we can use the job ID to identify and retrieve the results of the tasks. We also saw how we can use a flow chart to illustrate the logic of our API, and how we can test our API using curl or a browser.

I hope you enjoyed this post and learned something new. If you have any questions or feedback, please let me know in the comments. Thank you for reading! 😊

Sources:

FastAPI
Background Tasks - FastAPI

CodeWiseGuru

Search This Blog