Convert Your FastAPI App to Asynchronous

How to Convert Your FastAPI App to Asynchronous

In this tutorial, you will learn how to transition your existing synchronous FastAPI application to an asynchronous one. We will cover the fundamental differences between synchronous and asynchronous programming, understand when to leverage async routes, and systematically convert your entire application, including database interactions and exception handling, to utilize asynchronous operations. This process will make your application more efficient and capable of handling higher concurrency.

Understanding Synchronous vs. Asynchronous

Synchronous (sync) code executes tasks sequentially, one after another. Think of it like a sandwich shop where each customer’s order is completed entirely before the next customer is served. Asynchronous (async) code, on the other hand, allows your program to handle multiple tasks concurrently. It’s akin to a fast-food restaurant where an order can be taken, and while the food is being prepared, the staff can take the next order. This is particularly beneficial for I/O-bound tasks, such as database queries, network requests, or file operations, where the program would otherwise be idle, waiting for a response.

When to Use Async

• I/O-Bound Tasks: Ideal for operations that involve waiting for external resources like databases, APIs, or file systems.
• High Concurrency: Significantly improves performance when handling many requests simultaneously.

When NOT to Use Async

• CPU-Bound Tasks: Does not speed up heavy calculations or data processing that keeps the CPU busy.
• Simple Operations: For very simple or fast queries, the overhead of async machinery might not provide noticeable benefits and could even slightly slow things down.
• Synchronous Libraries: If your dependencies are strictly synchronous, you’ll need to wrap them or use alternatives.

FastAPI’s Handling of Sync and Async Routes

FastAPI intelligently handles both synchronous and asynchronous route definitions:

• Synchronous Routes (def): When you define a route using a regular def function, FastAPI automatically runs it in a separate thread pool. This prevents the function from blocking the main event loop, allowing other requests to be processed concurrently.
• Asynchronous Routes (async def): Routes defined with async def run directly in FastAPI’s main event loop. If these routes perform I/O operations, they must use await. Failure to do so will block the event loop, negatively impacting performance.

Using synchronous routes is not inherently slow or wrong; FastAPI manages them effectively. The choice depends on the nature of the task within the route.

Prerequisites

• A FastAPI application with existing CRUD operations (as built in previous tutorials).
• Python 3.7+
• SQLAlchemy ORM
• A basic understanding of Python’s async and await keywords.

Steps to Convert Your FastAPI App to Asynchronous

1. Install Asynchronous Dependencies

To enable asynchronous database operations with SQLite, you need an asynchronous driver. For SQLAlchemy, this is aiosqlite. If you are using PostgreSQL, you would typically use asyncpg.

Use your preferred package manager (e.g., pip or uv) to install the necessary library:

uv add aiosqlite

Additionally, ensure you have greenlet installed, as it’s required for SQLAlchemy’s async mode:

uv add greenlet

2. Update Database Configuration (`database.py`)

Modify your database configuration file to use asynchronous components from SQLAlchemy.

• Imports: Change imports from sqlalchemy to sqlalchemy.ext.asyncio for async-specific functionalities.

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# ... other imports

• Database URL: Append the async driver name to your SQLite URL.

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

• Engine Creation: Use create_async_engine instead of create_engine.

engine = create_async_engine(DATABASE_URL)

• Session Factory: Update sessionmaker to AsyncSession and configure it for asynchronous use. Setting expire_on_commit=False is recommended for async to prevent issues with expired objects after commits.

async_session_local = sessionmaker(
    class_=AsyncSession,
    expire_on_commit=False
)

• Get DB Function: Convert the get_db function into an asynchronous generator using async def and async with.

async def get_db() -> AsyncSession:
    async with async_session_local() as session:
        yield session

3. Update Main Application File (`main.py`)

Make the following changes in your main FastAPI application file:

• Imports: Add imports for asynccontextmanager, async exception handlers, and async session. Also import select and inload for eager loading.

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.exceptions import HTTPExceptionHandler, RequestValidationException
from sqlalchemy.orm import selectinload
from sqlalchemy.future import select
from sqlalchemy.ext.asyncio import AsyncSession

# ... other imports

• Lifespan Function: Replace the synchronous table creation with an asynchronous lifespan function. This function handles startup events (like creating tables) and shutdown events (like disposing of the engine).

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: Create tables
    async with engine.begin() as conn:
        # Use run_sync to execute sync create_all within async context
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Shutdown: Dispose of the engine
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

Expert Note: The lifespan function replaces older decorators like @app.on_event("startup") and @app.on_event("shutdown"). It provides a cleaner way to manage application lifecycle events.

4. Understand Eager Loading with Relationships

A crucial difference in async SQLAlchemy is the lack of support for lazy loading relationships. If you try to access a related object (e.g., post.author) without explicitly loading it, you will encounter an error. The solution is eager loading using selectinload.

You must explicitly tell SQLAlchemy to load related data when you fetch the primary object. This is done by adding .options(selectinload(Model.relationship)) to your select statements.

5. Convert Routes to Asynchronous

Iterate through your routes and apply the following changes:

• Change function definitions from def to async def.
• Change the dependency injection for the database session from the synchronous get_db to the asynchronous version.
• Add the await keyword before any database operations (e.g., db.execute(...), db.commit(), db.refresh(...), db.delete(...)).
• Important: db.add(...) does not require await as it’s an in-memory operation. However, db.commit() and db.refresh() do.
• Incorporate selectinload for any relationships that will be accessed after fetching the data.

Example: Converting the Home Route

@app.get("/")
async def home(db: AsyncSession = Depends(get_db)):
    # Use selectinload to eagerly load the author relationship
    posts = await db.execute(
        select(models.Post).options(selectinload(models.Post.author))
    )
    posts = posts.scalars().all()
    return templates.TemplateResponse("index.html", {"request": request, "posts": posts})

Expert Note: For routes that return API responses including related data, eager loading is essential. You can also use the attribute_names parameter with db.refresh() to load relationships after a commit, as shown in the create/update post routes.

Apply these changes consistently to all your routes that interact with the database. This includes GET, POST, PUT, PATCH, and DELETE operations.

Example: Create Post Route Snippet

@app.post("/posts")
async def create_post(
    post_data: schemas.PostCreate,
    db: AsyncSession = Depends(get_db),
    request: Request
):
    user = await db.get(models.User, post_data.user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    db_post = models.Post(
        title=post_data.title,
        content=post_data.content,
        user_id=post_data.user_id
    )
    db.add(db_post)  # No await needed for add
    await db.commit() # Await needed for commit
    await db.refresh(db_post, attribute_names=["author"])
    return templates.TemplateResponse("create_post.html", {"request": request, "post": db_post, "user": user})

6. Update Exception Handlers

Convert your custom synchronous exception handlers to asynchronous ones, leveraging FastAPI’s built-in async handlers for consistency.

• Change the handler function definitions to async def.
• Replace manual JSON responses with await calls to FastAPI’s default exception handlers (e.g., HTTPExceptionHandler, RequestValidationException).

# Example for HTTP Exception Handler
async def http_exception_handler(request: Request, exc: HTTPException):
    # Use default handler for API routes
    if request.url.path.startswith("/api"):
        return await solve_http_exception_api(request, exc)
    # Handle for templates
    return templates.TemplateResponse("error.html", {"request": request, "status_code": exc.status_code, "detail": exc.detail}, status_code=exc.status_code)

# Example for Request Validation Exception Handler
async def request_validation_exception_handler(request: Request, exc: RequestValidationException):
    if request.url.path.startswith("/api"):
        return await solve_request_validation_exception_api(request, exc)
    return templates.TemplateResponse("error.html", {"request": request, "status_code": 422, "detail": exc.errors()}, status_code=422)

# Register handlers (assuming solve_*_api are your async API handlers)
app.add_exception_handler(HTTPException, http_exception_handler)
app.add_exception_handler(RequestValidationException, request_validation_exception_handler)

7. Test Your Asynchronous Application

Run your FastAPI server and thoroughly test all endpoints, including creating, reading, updating, and deleting data. Verify that all functionalities work as expected, especially those involving relationships between models.

uvicorn main:app --reload

Check your application’s behavior through the browser and the automatic API documentation (Swagger UI). Ensure that related data is loaded correctly, confirming that your eager loading strategies are effective.

Conclusion and Best Practices

Converting your FastAPI application to asynchronous operations can significantly enhance its ability to handle concurrent loads, especially for I/O-bound tasks like API calls and database interactions.

• Mixed Approach: You don’t need to make all routes asynchronous. A mixed approach, where synchronous routes handle CPU-bound tasks or simpler operations and asynchronous routes handle I/O-bound tasks, is common and effective.
• Avoid Blocking IO: Never perform blocking I/O operations within an async def route without proper handling (e.g., using run_in_executor or libraries like httpx instead of requests).
• Performance Under Load: The most significant benefits of async are realized under high concurrency. For single, isolated requests, the performance difference might be negligible.
• Readability vs. Performance: Prioritize code clarity unless performance optimization is a critical requirement. Synchronous code is often easier to read and debug.

By following these steps, you have successfully transformed your FastAPI application to harness the power of asynchronous programming, making it more efficient and scalable.

---

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