Hackforge Academy

Explanation

Async programming allows a FastAPI application to handle other requests while waiting for an I/O operation such as a database query, HTTP request, file operation, or external API response. FastAPI is built on the ASGI standard and supports asynchronous endpoints using async and await. An async function is defined using async def, while await pauses that function until an asynchronous operation completes. This is especially useful for I/O-bound applications that handle many concurrent requests.

Code:

from fastapi import FastAPI
import asyncio

app = FastAPI()


@app.get('/hello')
async def hello():
    await asyncio.sleep(2)

    return {
        'message': 'Hello from async FastAPI'
    }


# Run:
# uvicorn main:app --reload

Explanation

The async def keyword creates an asynchronous function, and await is used to wait for another asynchronous operation. While the current request is waiting, the FastAPI server can work on other requests. This makes async programming particularly useful when an endpoint needs to call external APIs, databases, or other microservices. However, simply adding async to a function does not automatically make blocking operations asynchronous. The operations being awaited should also support asynchronous execution.

Code:

import asyncio
from fastapi import FastAPI

app = FastAPI()


async def get_data():
    await asyncio.sleep(2)
    return 'Data received'


@app.get('/data')
async def data():
    result = await get_data()

    return {
        'result': result
    }

Explanation

One of the most common uses of asynchronous programming in FastAPI is calling another REST API or microservice. The httpx library provides AsyncClient for making asynchronous HTTP requests. While FastAPI waits for the external service to respond, the server can continue handling other requests. Timeouts should always be configured for external calls so that a slow or unavailable service does not cause requests to wait indefinitely.

Code:

from fastapi import FastAPI
import httpx

app = FastAPI()


@app.get('/products')
async def get_products():
    async with httpx.AsyncClient() as client:
        response = await client.get(
            'https://api.example.com/products',
            timeout=5.0
        )

    return response.json()

Explanation

When an endpoint needs to call multiple independent services, asyncio.gather() can execute asynchronous operations concurrently. Instead of waiting for Service A to finish before calling Service B, both operations can be started together. This can significantly reduce total response time when the operations are independent and I/O-bound. This pattern is useful in microservices applications where an API may need data from multiple services.

Code:

import asyncio
import httpx
from fastapi import FastAPI

app = FastAPI()


async def get_users(client):
    response = await client.get(
        'https://api.example.com/users'
    )
    return response.json()


async def get_products(client):
    response = await client.get(
        'https://api.example.com/products'
    )
    return response.json()


@app.get('/dashboard')
async def dashboard():
    async with httpx.AsyncClient(timeout=5.0) as client:
        users, products = await asyncio.gather(
            get_users(client),
            get_products(client)
        )

    return {
        'users': users,
        'products': products
    }

Explanation

Async programming works best for I/O-bound operations such as HTTP calls, asynchronous database operations, file or network operations, and message queues. Avoid blocking the event loop with long-running synchronous operations such as time.sleep(), heavy CPU calculations, or blocking database calls inside an async endpoint. For blocking work, use appropriate synchronous endpoints, thread pools, background workers, or separate services. In production FastAPI applications, async programming should be combined with timeouts, exception handling, connection pooling, logging, and proper resource management.

Code:

import asyncio
from fastapi import FastAPI

app = FastAPI()


# Good: asynchronous waiting
@app.get('/good')
async def good_example():
    await asyncio.sleep(2)
    return {'message': 'Async operation completed'}


# Avoid blocking the event loop:
# import time
# time.sleep(2)
#
# For CPU-heavy tasks, consider:
# - Background workers
# - Task queues
# - Separate services
# - Process-based execution