Moving from Sync to Async in FastAPI with SQLModel—What you Need to Know

This one's technical.

When starting with database connections, most tutorials focus on synchronous setups. While this approach is simpler to implement, it can become a bottleneck as your application scales, especially under high load or with long-running queries. To handle increased traffic and improve performance, migrating to an asynchronous (async) database connection is often the next step. This article will guide you through the process of transitioning from a synchronous to an asynchronous setup in FastAPI/SQLModel, highlighting key considerations and challenges.

Synchronous vs. Asynchronous Execution

In a synchronous setup, tasks are executed sequentially. Each database request must wait for the previous one to complete before it can begin. This can lead to performance bottlenecks, particularly in high-concurrency scenarios or when dealing with slow I/O operations like database queries or file uploads.

In contrast, asynchronous execution allows multiple tasks to be processed concurrently. Instead of waiting for I/O operations (e.g., database calls) to finish, the application can handle other tasks in the meantime. This approach maximizes resource utilization, reduces idle time, and significantly improves performance in high-concurrency environments.

Switching from a synchronous setup to an asynchronous one in your FastAPI/SQLModel app can bring a variety of challenges. When making this switch, several important details need attention, especially around async sessions, avoiding misuse of await, and correctly handling relationships between tables.

A Typical Synchronous Setup

This code creates the synchronous DB engine, connected to your DB, here postgres. Then, you create the session in which you'll make your DB requests, bound to the engine you just created. Using this setup, your DB calls will all be handled sequentially.



# sync version
engine = create_engine(pg_connection_string)
session = sessionmaker(bind=engine)

def get_sync_session():
    with session() as db_session:
        yield db_session

This works well in synchronous applications but falls short when scaling up to asynchronous systems. Let's learn how to setup FastAPI to operate async.

Solution

Step 1: Set up an Async Session

Instead of using the synchronous create_engine, you’ll use create_async_engine to establish your async connection. Likewise, you'll use the async_sessionmaker:



from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlmodel import SQLModel
from typing import AsyncGenerator
from shared.config import database_settings

# Step 1: Create async engine and session
async_engine = create_async_engine(
    async_pg_connection_string,  # Async connection string
    echo=True,  # Optional: Set to False in production
    future=True,
)

# Step 2: Set up async session
async_session = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

# Step 3: Create a dependency to yield an async session
async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session

Important Note: You’ll often see



async def get_async_session() -> AsyncGenerator[AsyncSession, None]:

across the internet. This pattern yields an async generator used in FastAPI's dependency injection system. It’s not the same as simply doing



async with async_session() as session:

which only creates a context manager. So, keep it simple as use the async_session() directly when called for.

Step 2: Use the Async Session in Your Routes

Once the async session is set up, update your routes to use it via Depends:



from fastapi import Depends

@router.post("/search")
async def query_stuff(  # <-- mark the function as async
    query_params: QueryParams = Body(...),
    session: AsyncSession = Depends(get_async_session),  # <-- use async session
):
    # Perform async DB operations here
    result = await session.execute(query)  # <-- async db operation
    return result.fetchall()

Instead of just calling result = session.execute(query) we have to await that call.

Step 3: Do Not Overuse await on Session Methods

One common mistake when switching to async is to start adding await to every session method call. You should only use await when calling truly asynchronous operations, such as:

Queries like session.execute()
Commit and rollback operations like session.commit()

For example:



# Correct usage
result = await session.execute(query)
await session.commit()

However, do not add await to actions like creating a session or other methods that aren't actually async:



# Incorrect: do not add await to session() calls
await session  # <-- This will raise an error!

Only add await to functions that require IO-bound operations such as querying the database or making external API calls.

Step 4: Managing Relationships with Care

When you move to async, managing relationships between tables becomes a real concern. If you lazily load related models by using



"lazy": "selectin"

without understanding the implications, you may accidentally trigger IO-on-attribute access (i.e., issuing a query on the database as soon as a related attribute is accessed). This can cause greenlet-related errors, which can be frustrating to debug.

Huh? When you load data from a relational DB you often end up joining tables to pull in the related data. For example if I want to call person.address I'll end up joining the address and person tables. With lazy loading, the person data will be loaded, then a second call will be made to fetch the address data. It's this second call that will hose you; it's the "IO-on-attribute access".

You can work around this by eagerly loading the data (eg. address is automatically loaded with person). Carefully consider when and how related models should be loaded. For example, using



"lazy": "joined"

ensures the related data is fetched eagerly in a single query:



from typing import List
from sqlmodel import Relationship

class User(SQLModel, table=True):
    id: int
    name: str
    oauth_accounts: List["OAuthAccount"] = Relationship(
        back_populates="user",
        sa_relationship_kwargs={"lazy": "joined", "cascade": "all, delete"},
    )

Here, the "lazy": "joined" strategy ensures that whenever a User is queried, the related OAuthAccount records are pulled in a single query. This prevents unexpected IO on attribute access that can lead to performance bottlenecks or errors in async contexts.

Why It Matters

As mentioned before, lazy-loading relationships can lead to unexpected database IO during attribute access, which can cause performance degradation and errors in async applications. By carefully choosing when to join related tables (using "lazy": "joined"), you ensure better performance and avoid IO-related greenlet issues.

Step 5: Handling Transactions and Sessions Properly

When moving to async, the way sessions are managed also changes. Make sure you:

Use expire_on_commit=False: This ensures that once a session is committed, you can still access the data without the need for an additional query. This is important in async environments where additional queries might trigger IO unexpectedly.
Always await on commit() and rollback(): Database operations like committing a transaction are IO-bound and must be awaited.



# Async transaction example
async def create_user(session: AsyncSession, user_data: UserCreate) -> User:
    new_user = User(**user_data.dict())
    session.add(new_user)
    await session.commit()  # Always await commits
    return new_user

Wrapping Up

Moving from sync to async in FastAPI with SQLModel can significantly boost performance but requires careful management of sessions, relationships, and async operations. By setting up an async session properly, avoiding misuse of await, and handling relationships thoughtfully, you can avoid common pitfalls like connection exhaustion and unexpected IO.

Happy coding!