Skip to main content
Back to path

Welcome to a Grasp lesson

Create a free account to keep studying and create your own study path.

Lesson 43

Building High-Concurrency APIs with FastAPI Async Handlers

Recommended resources

Welcome to Module 7! In our previous module, we focused on building robust, high-performance gRPC services, culminating in a lesson where you implemented interceptors to handle cross-cutting concerns like logging and authentication. Our gRPC service is now secure and well-structured, but it's designed for internal, service-to-service communication.

This module shifts our focus to the "front door" of our microservices system: the API Gateway. This is the public-facing entry point that clients like web or mobile apps will interact with. For this, we'll be using FastAPI, a modern, high-performance Python web framework built for exactly this purpose.

In this first lesson, you will learn how to implement the core of any fast API: high-concurrency request handlers. We'll dive deep into how FastAPI leverages Python's asyncio to handle many requests simultaneously, a critical foundation for building a gateway that can scale to meet high traffic demands.

1. FastAPI's Concurrency Model: async def vs. def

One of FastAPI's most powerful features is its ability to handle both standard synchronous functions (defined with def) and asynchronous coroutines (defined with async def) as your API endpoints. Understanding how it treats them differently is the key to unlocking high performance.

  • async def: When you declare a path operation function with async def, FastAPI runs it directly on the main event loop. This is extremely efficient for I/O-bound operations (like calling a database, another microservice, or reading a file) because it can pause the function while it waits, and use that time to work on other incoming requests.
  • def: If you use a regular def, FastAPI is smart enough to know that it might contain blocking I/O code. To prevent this from freezing the entire application, it runs the function in a separate thread from a thread pool. This provides concurrency but comes with the overhead of managing threads.

To see this in action, the following video provides a crystal-clear demonstration.

How FastAPI Handles Requests Behind the Scenes

This short video from Code Collider provides an excellent practical demonstration of how FastAPI handles blocking and non-blocking code in both async def and def endpoints.

Watch the entire video. Pay close attention to the terminal output in these three experiments: An async def endpoint with a blocking time.sleep(). Notice how requests are processed one after the other. An async def endpoint with a non-blocking asyncio.sleep(). See how FastAPI starts processing the second request while the first is "sleeping". A def endpoint with a blocking time.sleep(). Observe that the requests are handled concurrently, but for a different reason. The explanation starting at "When you run fastapi application using uvicorn..." perfectly summarizes the underlying mechanics of the event loop and the thread pool.

The main takeaway is a fundamental rule for writing high-concurrency services in FastAPI: Never call blocking I/O code inside an async def function. Doing so will block the entire event loop, defeating the purpose of using asyncio and grinding your application to a halt under load.

FastAPI Concurrency with Async/Await
This diagram shows the ideal scenario for an `async def` endpoint. Multiple requests are received and handled as concurrent tasks by the event loop, leading to high throughput.
Trap #1: Blocking Call Inside async def
This diagram illustrates the critical mistake of using a synchronous, blocking database call inside an `async def` endpoint. The blocking call freezes the entire event loop, forcing all other concurrent requests to wait.

2. Hands-On: From Synchronous to Asynchronous API

Theory is one thing, but converting a real application is the best way to learn. You are about to refactor a simple, database-backed synchronous FastAPI application to be fully asynchronous, enabling it to handle much higher concurrency.

The following comprehensive video tutorial by Corey Schafer will guide you through every step of this process. It covers the exact challenges you'd face in a real-world project. We will break it down into manageable steps.

Python FastAPI Tutorial (Part 7): Sync vs Async - Converting Your App to Asynchronous

This video provides a conceptual introduction to asynchronous programming, its benefits for I/O-bound tasks, and how FastAPI handles both async and sync code.

Start by watching the first four minutes to solidify your understanding. Focus on: The analogy of McDonald's vs. Subway for async vs. sync execution. The distinction between I/O-bound vs. CPU-bound tasks. Our database calls are I/O-bound, making them perfect for asyncio. How FastAPI intelligently handles def vs. async def, which reinforces the previous video.

Now, let's get our hands dirty. The rest of the video walks through a practical conversion.

Step 1: Async Database Configuration

To speak to a database asynchronously, we need three things: an async database driver, an async SQLAlchemy engine, and an async session maker.

Python FastAPI Tutorial (Part 7): Sync vs Async - Converting Your App to Asynchronous

This section shows how to set up the foundational database components for async operations.

Watch the segment from installing dependencies. Key changes to observe are: Installing an async driver (aiosqlite in this case; for PostgreSQL, it would be asyncpg). Switching from SQLAlchemy's standard imports to those from sqlalchemy.ext.asyncio (e.g., create_async_engine, async_session_maker). Updating the get_db dependency to be an async def that yields an AsyncSession.

Step 2: Async Application Lifespan

We can't just call the synchronous Base.metadata.create_all() at the module level anymore. We need to run it in an async-friendly context when the application starts. FastAPI provides the lifespan context manager for this.

Python FastAPI Tutorial (Part 7): Sync vs Async - Converting Your App to Asynchronous

This part covers the modern way to handle startup and shutdown events in FastAPI.

Watch the section explaining the lifespan function. Note how engine.begin() is used to get an async connection and conn.run_sync() is used to execute the synchronous create_all method within an async context.

Step 3: The Challenge of Lazy Loading

This is one of the most important and subtle differences when working with async SQLAlchemy. In a synchronous app, SQLAlchemy can "lazy load" related objects; when you access post.author, it automatically runs a new query in the background. This does not work in async. Lazy loading would require a blocking database call, which would freeze the event loop.

The solution is eager loading. We must explicitly tell SQLAlchemy which relationships to load in our initial query.

Python FastAPI Tutorial (Part 7): Sync vs Async - Converting Your App to Asynchronous

This section explains the critical concept of eager loading, which is mandatory in async SQLAlchemy.

Watch from "I need to explain something critical...". Absorb the explanation of why lazy loading fails and how selectinload is used to solve this by loading related objects upfront.

Step 4: Converting the API Endpoints

With the foundation in place, we can now convert our endpoint functions. The pattern is consistent and straightforward.

Python FastAPI Tutorial (Part 7): Sync vs Async - Converting Your App to Asynchronous

This is where all the pieces come together as the synchronous endpoints are refactored.

Watch the final segments on converting routes and exception handlers. You don't need to type everything, but focus on recognizing the pattern: Change def to async def. Add await before any database operation: session.execute, session.commit, session.refresh. Add .options(selectinload(models.Post.author)) to queries where a relationship will be accessed later. Notice that session.add() does not need await because it only manipulates the session in memory.

3. A Complete Example

Here is a simplified code snippet that brings together all the concepts you've just seen. It shows a single async endpoint that fetches data from a database, correctly using an async session and awaiting the I/O operation.




# main.py
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from contextlib import asynccontextmanager




# Assume database.py and models.py are set up as shown in the video
from . import database, models




# Lifespan function for async startup/shutdown
@asynccontextmanager
async def lifespan(app: FastAPI):



    # On startup: Create tables
    async with database.engine.begin() as conn:
        await conn.run_sync(models.Base.metadata.create_all)
    yield



    # On shutdown: Dispose of the engine connection pool
    await database.engine.dispose()

app = FastAPI(lifespan=lifespan)




# Async endpoint for fetching posts
@app.get("/posts")
async def get_posts(db: AsyncSession = Depends(database.get_db)):
    """
    Fetches all posts from the database asynchronously.
    """



    # Build the query
    query = select(models.Post)
    



    # Execute the query and await the result (I/O operation)
    result = await db.execute(query)
    



    # .scalars().all() is a synchronous operation on the in-memory result
    posts = result.scalars().all()
    
    return posts

By defining the get_posts endpoint with async def and using await for the database call, this endpoint can now handle many concurrent requests efficiently. While one request is waiting for the database to respond, FastAPI's event loop will be busy processing other requests, drastically improving the throughput of your application.

Conclusion

In this lesson, you took a significant step toward building a high-performance API gateway. You've learned the fundamental difference between how FastAPI handles synchronous and asynchronous code and, most importantly, you've walked through the practical process of converting a database-driven application to be fully async.

Key Takeaways:

  • async def for I/O-bound tasks: Use async def for endpoints that perform network calls or database queries to allow the event loop to handle requests concurrently.
  • def for blocking/CPU-bound tasks: FastAPI runs def endpoints in a thread pool, preventing them from blocking the event loop. This is a safe fallback but less efficient for I/O than a true async implementation.
  • Never block the event loop: Calling synchronous, blocking code (like time.sleep() or a sync database driver) inside an async def function is the number one performance killer.
  • Async database integration requires specific tools: You must use an async driver (e.g., asyncpg), create_async_engine, AsyncSession, and await all database operations.
  • Eager loading is mandatory: Lazy loading is not compatible with asyncio. Use selectinload to explicitly fetch related data in your initial query.

Preview of the Next Lesson

In our example, we used db: AsyncSession = Depends(database.get_db). This Depends function is a key part of FastAPI's Dependency Injection system. In the next lesson, you will take a deep dive into this system to learn how to manage shared resources like database connections, configurations, and authenticated user objects in a clean, robust, and testable way.