Moving from Sync to Async in FastAPI with SQLModel? Here's What You Need to Know!

Switching from a synchronous setup to an asynchronous one in your FastAPI/SQLModel app can bring some 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:

# sync version
engine = create_engine(database_settings.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.

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:

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(
    database_settings.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() direclty 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(
    query_params: QueryParams = Body(...),
    session: AsyncSession = Depends(get_async_session),  # Use async session
):
    # Perform async DB operations here
    result = await session.execute(query)
    return result.fetchall()

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.

To avoid this, 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:

Lazy-loading relationships can lead to unexpected database IO during attribute access, which can cause performance degradation 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

Final Thoughts:

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!

inchoate/moving-to-fastapi-async.md