FastAPI with Async Request and Response: A Comprehensive Guide

FastAPI is a modern, high-performance web framework for building APIs with Python 3.7+ based on standard Python type hints. One of its standout features is its native support for asynchronous programming, enabling developers to handle high-throughput and concurrent requests efficiently.

In this blog, we will delve into how to use FastAPI for asynchronous requests and responses, providing detailed explanations and practical examples.

Why Async in FastAPI?

Asynchronous programming allows for non-blocking operations, making it ideal for I/O-bound tasks such as:

🠮 Calling external APIs.
🠮 Interacting with databases.
🠮 Performing file I/O operations.

Using async in FastAPI can significantly improve the scalability and responsiveness of your API.

Related read: An Introduction to FastAPI, a Powerful Python Framework

Setting Up FastAPI

First, install FastAPI and an ASGI server such as Uvicorn:

pip install fastapi uvicorn

Create a file called main.py to start building your FastAPI app.

Example: A Simple Asynchronous Endpoint

Here is a basic example of an asynchronous endpoint:

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/async-endpoint")
async def async_endpoint():
    await asyncio.sleep(2) # Simulate a time-consuming operation
    return {"message": "This is an async response!"}

Key Points

🠮 “: Defines the endpoint as an asynchronous function.
🠮 “: Pauses execution of the function until the awaited task is complete.

When you hit this endpoint, FastAPI will handle other incoming requests during the 2-second delay, demonstrating concurrency.

Example: Making Async API Calls

FastAPI works seamlessly with httpx, a library designed for making asynchronous HTTP requests.

Install httpx:

pip install httpx

Example Code:

from fastapi import FastAPI
import httpx


app = FastAPI()


@app.get("/fetch-data")
async def fetch_data():
      async with httpx.AsyncClient() as client:
      response = await client.get("https://jsonplaceholder.typicode.com/posts")
      return response.json()

Key Points:

🠮 “: Used for making non-blocking HTTP requests.
🠮 “: Waits for the response while freeing the event loop to handle other tasks.

Example: Asynchronous Database Operations

Asynchronous database interactions are critical for performance in data-heavy applications. You can use libraries like databases or SQLAlchemy (async version).

Install databases:

pip install databases

Example Code:

from fastapi import FastAPI
from databases import Database


DATABASE_URL = "sqlite+aiosqlite:///./test.db"


database = Database(DATABASE_URL)
app = FastAPI()


@app.on_event("startup")
async def startup():
      await database.connect()


@app.on_event("shutdown")
async def shutdown():
      await database.disconnect()


@app.get("/users")
async def get_users():
      query = "SELECT * FROM users"
      results = await database.fetch_all(query)
      return results

Key Points:

🠮 “: Executes a non-blocking query.
🠮 Lifecycle Events: @app.on_event(“startup”) and @app.on_event(“shutdown”) manage the database connection.

Understanding Concurrency in FastAPI

FastAPI leverages Python’s asyncio for concurrency. Here’s an example to demonstrate concurrent tasks:

from fastapi import FastAPI
import asyncio


app = FastAPI()


@app.get("/parallel-tasks")
async def parallel_tasks():
async def task_1():
    await asyncio.sleep(2)
    return "Task 1 complete"


async def task_2():
    await asyncio.sleep(3)
    return "Task 2 complete"


    results = await asyncio.gather(task_1(), task_2())
    return {"results": results}

Key Points:

🠮 “: Runs multiple coroutines concurrently.
🠮 The total execution time here will be approximately 3 seconds, as the tasks run in parallel.

Learn More About Our Python Development Expertise

Error Handling with Async Endpoints

FastAPI provides a robust mechanism for handling errors, even in asynchronous endpoints.

Example Code:

from fastapi import FastAPI, HTTPException
import httpx


app = FastAPI()


@app.get("/external-api")
async def external_api():
      try:
      async with httpx.AsyncClient() as client:
            response = await client.get("https://nonexistent-api.com/data")
            response.raise_for_status()
      except httpx.HTTPStatusError as e:
      raise HTTPException(status_code=e.response.status_code, detail="Error fetching data")
      except httpx.RequestError:
      raise HTTPException(status_code=500, detail="Request failed")

      return response.json()

Performance Benefits of Async

1. High Concurrency: It handles thousands of requests simultaneously without blocking.
2. Efficient I/O: Ideal for applications with many external API calls or database operations.
3. Scalability: Reduces the need for additional threads, leading to lower memory usage.

Testing Async Endpoints

You can test async endpoints using pytest and httpx.

Install Pytest and Testclient:

pip install pytest httpx

Example Test:

from fastapi.testclient import TestClient
from httpx import AsyncClient
from main import app


client = TestClient(app)


async def test_async_endpoint():
      async with AsyncClient(app=app, base_url="http://test") as ac:
      response = await ac.get("/async-endpoint")
      assert response.status_code == 200
      assert response.json() == {"message": "This is an async response!"}
coma

Conclusion

FastAPI’s async support enables developers to build highly efficient and scalable APIs. By leveraging Python’s asyncio, FastAPI allows for non-blocking I/O, making it an excellent choice for modern web applications requiring high concurrency.

With the examples provided, you can now confidently build APIs with asynchronous request and response handling in FastAPI. Experiment with various use cases, such as external API calls and database operations, to unlock the full potential of asynchronous programming!

Keep Reading

Keep Reading

  • Service
  • Career
  • Let's create something together!

  • We’re looking for the best. Are you in?