All-in-one Collaborative API Development Platform

API Design

API Documentation

API Designing

API Mock

API Automated Testing

Sign up for free

FastAPI Threading Made Easy: Improving Concurrency in Python

Start for free
Home / Basic Knowledge / FastAPI Threading Made Easy: Improving Concurrency in Python

FastAPI Threading Made Easy: Improving Concurrency in Python

This article will explore the usage of multi-threading in the FastAPI framework, covering common use cases, problem-solving approaches, and practical examples.

In modern web applications, high performance and fast responsiveness are crucial. Python's FastAPI framework has become the preferred choice for many developers due to its outstanding performance and user-friendly features.

In Python, a thread is a separate flow of execution that can run concurrently with other threads. Each thread operates independently, and multiple threads can be created and executed within a single program. This is in contrast to the traditional sequential execution, where tasks are executed one after the other.

FastAPI thread

In certain scenarios, single-threaded execution may not meet the requirements, prompting the consideration of using multi-threading to enhance application concurrency. This article will explore the usage of multi-threading in the FastAPI framework, covering common use cases, problem-solving approaches, and practical examples.

Multi-threading in Python

One of the main benefits of multi-threading is improved performance. By dividing a task into multiple threads, each responsible for a specific subtask, the overall execution time can be reduced. This is particularly useful in FastAPI applications, where tasks such as handling requests, processing data, and performing computations can be parallelized.

To understand multi-threading in Python, it's essential to grasp some key concepts. First, Python's Global Interpreter Lock (GIL) restricts the execution of multiple threads simultaneously in the same process. This means that even though multiple threads are created, only one thread can execute Python bytecode at a time. As a result, multi-threading in Python does not provide true parallelism but rather concurrent execution.

However, multi-threading can still offer performance improvements in FastAPI applications. While the GIL prevents multiple threads from executing Python code simultaneously, it does not restrict other operations, such as I/O operations, network requests, or database queries. By offloading these operations to separate threads, the main thread can focus on other tasks, effectively utilizing the available resources and improving overall performance.

In FastAPI, implementing multi-threading involves creating and managing threads to perform specific tasks concurrently. This can be achieved using the threading module in Python, which provides a high-level interface for creating and managing threads. By creating and starting multiple threads, developers can distribute the workload across different threads and improve the application's responsiveness.

How to Use Multi-Threading in FastAPI?

Step 1. Creating Threads in Path Operation Functions

You can directly create threads using the threading module within path operation functions.

import threading
from fastapi import FastAPI

app = FastAPI()

def get_resource():
    t = threading.Thread(target=do_work) 
    return {"message": "Thread started"}

def do_work():
    # perform computationally intensive work here

Step 2. Utilizing Background Tasks

FastAPI provides the @app.on_event("startup") decorator to create background tasks during startup.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def startup_event():
    threading.Thread(target=do_work, daemon=True).start() 

def do_work():
    while True:
       # perform background work

Step 3. Using Third-Party Background Task Libraries

You can employ third-party libraries like apscheduler to execute periodic background tasks.

from apscheduler.schedulers.background import BackgroundScheduler
from fastapi import FastAPI

app = FastAPI()
scheduler = BackgroundScheduler()

def start_background_processes():
    scheduler.add_job(do_work, "interval", seconds=5)

def do_work():
    # perform periodic work

Practice Example

Installing Required Tools
First, ensure that you have Python and pip installed. Then, use the following command to install FastAPI and uvicorn:

pip install fastapi 
pip install uvicorn

Writing a Multi-threaded Application
Let's create a simple FastAPI application:

from fastapi import FastAPI
import threading
import time

app = FastAPI()

# Time-consuming task function
def long_running_task(task_id: int):
    print(f"Starting Task {task_id}")
    time.sleep(5)  # Simulate task execution time
    print(f"Finished Task {task_id}")

# Background task function
def run_background_task(task_id: int):
    thread = threading.Thread(target=long_running_task, args=(task_id,))

async def read_root():
    return {"Hello": "World"}

async def run_task_in_background(task_id: int):
    # Create and start a background thread to run the task
    return {"message": f"Task {task_id} is running in the background."}

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="", port=8000)

In this example, we use the threading.Thread class to create a background thread for running the time-consuming task. Inside the run_background_task function, we create a thread object thread with long_running_task as its target function, passing task_id as an argument. Then, we call thread.start() to launch a new thread and run long_running_task in the background.

Running the Application
Use the following command to run the FastAPI application:

uvicorn main:app --reload

Now, you can test the application using a browser or a tool like curl. For instance, open a browser, type http://localhost:8000/task/1, and hit Enter. You will immediately receive the response {"message": "Task 1 is running in the background."}, and the task running in the background will print "Finished Task 1" after 5 seconds. Meanwhile, you can continue sending other task requests.


Debugging FastAPI API with Apidog

Apidog is an all-in-one API collaboration platform that integrates API documentation, API debugging, API mocking, and API automation testing. With Apidog, we can easily debug FastAPI interfaces.

Apidog - An integrated platform for API design, debugging, development, mock, and testing
REAL API Design-first Development Platform. Design. Debug. Test. Document. Mock. Build APIs Faster & Together.

To quickly debug an API endpoint, create a new project and select the "Debug Mode." Fill in the request address, and you can swiftly send the request to obtain the response results, as illustrated in the example above.

Tips for FastAPI Multi-Threading

Multi-threading can be a powerful tool for optimizing code execution, but it also comes with its own set of challenges and considerations.

  1. Identify CPU-bound vs I/O-bound tasks: Before implementing multi-threading, it is important to identify the nature of the tasks in your FastAPI application. CPU-bound tasks are those that heavily rely on computational power, while I/O-bound tasks involve waiting for external resources like databases or APIs. Multi-threading is most effective for I/O-bound tasks, as it allows the application to perform other operations while waiting for the I/O operations to complete.
  2. Use an appropriate number of threads: The number of threads used in your FastAPI application should be carefully chosen. While it may be tempting to use a large number of threads, excessive thread creation can lead to resource contention and decreased performance. It is recommended to perform load testing and experimentation to determine the optimal number of threads for your specific application.
  3. Avoid shared mutable state: Multi-threading introduces the potential for race conditions and data inconsistencies when multiple threads access shared mutable state. To avoid such issues, it is important to design your FastAPI application in a way that minimizes shared mutable state. Use thread-safe data structures and synchronization mechanisms like locks or semaphores when necessary.
  4. Consider the Global Interpreter Lock (GIL): Python's Global Interpreter Lock (GIL) ensures that only one thread can execute Python bytecode at a time. This means that multi-threading in Python may not provide true parallelism for CPU-bound tasks. However, the GIL is released during I/O operations, allowing multi-threading to be effective for I/O-bound tasks. Consider using multiprocessing instead of multi-threading for CPU-bound tasks if parallelism is crucial.
  5. Implement error handling and graceful shutdown: When using multi-threading in FastAPI, it is important to handle errors gracefully and ensure proper shutdown of threads. Unhandled exceptions in threads can cause your application to crash or behave unexpectedly. Implement error handling mechanisms and use tools like concurrent.futures.ThreadPoolExecutor to gracefully shut down threads when they are no longer needed.
  6. Monitor and optimize thread performance: It is crucial to monitor the performance of your multi-threaded FastAPI application and identify any bottlenecks or areas for improvement. Use tools like profilers and performance monitoring libraries to identify areas of high CPU usage or excessive thread contention. Optimize your code and thread usage based on the insights gained from monitoring.