Supercharge Your API Performance with Asynchronous Programming in FastAPI

Ready to take your API game to the next level?

FastAPI is here to make your APIs faster, more responsive, and capable of handling heavy loads like a pro.

In this article, we’ll show you how to leverage asynchronous programming in FastAPI to build high-performance APIs. By the end, you’ll be equipped with the knowledge to implement async endpoints and test them effectively.

What You’ll Learn

Here’s what you’ll master:

• The basics of asynchronous programming and why it matters.
• How to set up a FastAPI environment for async development.
• Writing and testing async endpoints with real-world examples.
• Using async libraries for HTTP requests, file handling, and background tasks.

Why Use Asynchronous Programming in FastAPI?

What Is It?

Asynchronous programming enables tasks to run concurrently. This is particularly useful for tasks like network requests, database queries, or file operations where waiting for a response is common.

Why Does It Matter?

In traditional synchronous programming, tasks run sequentially, leading to delays when handling multiple requests. With asynchronous programming, you can serve multiple users simultaneously, maximizing resource utilization and ensuring a better user experience.

Setting Up Your FastAPI Environment

Now, let’s roll up our sleeves and build something amazing!

First, install the required libraries:

pip install "fastapi[standard]" httpx aiofiles pytest

The Code

Below is a complete example demonstrating asynchronous programming in FastAPI. Each part of the code serves a unique purpose, and we’ll explain them step by step.

from fastapi import FastAPI, BackgroundTasks
import httpx
import aiofiles
import pytest
from fastapi.testclient import TestClient

app = FastAPI()

# API Endpoints
@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

@app.get("/external-api")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
    return response.json()

@app.get("/read-file")
async def read_file():
    async with aiofiles.open("example.txt", mode="r") as file:
        content = await file.read()
    return {"content": content}

def send_email(email: str, message: str):
    print(f"Sending email to {email} with message: {message}")

@app.post("/send-email/")
async def schedule_email(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Email scheduled"}

# Testing Code
client = TestClient(app)

def test_read_item():
    response = client.get("/item/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

def test_read_file():
    # Create an example file for testing
    with open("example.txt", "w") as file:
        file.write("This is a test file content")

    response = client.get("/read-file")
    assert response.status_code == 200
    assert response.json() == {"content": "This is a test file content"}

def test_schedule_email():
    response = client.post("/send-email/?email=fogigav197@rabitex.com")
    assert response.status_code == 200
    assert response.json() == {"message": "Email scheduled"}

@pytest.mark.asyncio
async def test_call_external_api():
    async with httpx.AsyncClient() as async_client:
        response = await async_client.get("https://jsonplaceholder.typicode.com/posts/1")
    assert response.status_code == 200
    assert "id" in response.json()

Breaking It Down

API Endpoints

1.Simple Async Endpoint

@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

This endpoint showcases how to define a basic asynchronous route using async def. It retrieves an item by its ID.

2.Calling an External API

@app.get("/external-api")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
    return response.json()

This demonstrates how to make a non-blocking HTTP request using httpx. While waiting for the external API response, your application can process other requests.

3.Asynchronous File Reading

@app.get("/read-file")
async def read_file():
    async with aiofiles.open("example.txt", mode="r") as file:
        content = await file.read()
    return {"content": content}

This reads a file asynchronously, ensuring the file operation doesn’t block the application.

4.Background Task Execution

def send_email(email: str, message: str):
    print(f"Sending email to {email} with message: {message}")

@app.post("/send-email/")
async def schedule_email(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Email scheduled"}

This endpoint schedules a background task to send an email, allowing the main thread to handle other requests.

Testing the Code

1.Testing Basic Endpoint

def test_read_item():
    response = client.get("/item/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

This verifies the /item/{item_id} endpoint returns the expected data.

2.Testing File Reading

pip install "fastapi[standard]" httpx aiofiles pytest

This creates a test file and checks if the /read-file endpoint reads and returns its content correctly.

3.Testing Background Task

from fastapi import FastAPI, BackgroundTasks
import httpx
import aiofiles
import pytest
from fastapi.testclient import TestClient

app = FastAPI()

# API Endpoints
@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

@app.get("/external-api")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
    return response.json()

@app.get("/read-file")
async def read_file():
    async with aiofiles.open("example.txt", mode="r") as file:
        content = await file.read()
    return {"content": content}

def send_email(email: str, message: str):
    print(f"Sending email to {email} with message: {message}")

@app.post("/send-email/")
async def schedule_email(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Email scheduled"}

# Testing Code
client = TestClient(app)

def test_read_item():
    response = client.get("/item/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

def test_read_file():
    # Create an example file for testing
    with open("example.txt", "w") as file:
        file.write("This is a test file content")

    response = client.get("/read-file")
    assert response.status_code == 200
    assert response.json() == {"content": "This is a test file content"}

def test_schedule_email():
    response = client.post("/send-email/?email=fogigav197@rabitex.com")
    assert response.status_code == 200
    assert response.json() == {"message": "Email scheduled"}

@pytest.mark.asyncio
async def test_call_external_api():
    async with httpx.AsyncClient() as async_client:
        response = await async_client.get("https://jsonplaceholder.typicode.com/posts/1")
    assert response.status_code == 200
    assert "id" in response.json()

This tests whether the background email task is successfully scheduled.

4.Testing External API Call

@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

This ensures the /external-api endpoint correctly fetches data from an external source.

Output

Conclusion

With the provided code, you now have a practical understanding of how to build and test async APIs using FastAPI. Whether it’s handling files, calling external APIs, or scheduling background tasks, asynchronous programming lets you create high-performance applications that scale effortlessly.

Ready to build your next FastAPI project? Let’s get started!

Thanks for reading...
Happy Coding!

The above is the detailed content of Supercharge Your API Performance with Asynchronous Programming in FastAPI. For more information, please follow other related articles on the PHP Chinese website!