Python FastAPI Async SQLAlchemy Database Setup

The Blocking Problem with Synchronous Sessions

When you combine python fastapi sqlalchemy async database, the first thing to understand is why a synchronous SQLAlchemy session cannot be used directly inside an async endpoint. A sync session performs blocking I/O when it talks to the database. If you call that session from an async route handler, the event loop stalls while the database query runs. Other requests that are perfectly ready to proceed wait for that blocking call to finish. The result is an application that behaves like a synchronous server, even though it is built on FastAPI's async stack.

The fix is to use SQLAlchemy's async extension, which provides an async engine and an async session that yield control back to the event loop during database I/O. SQLAlchemy 2.0 ships this support as part of the core library, so you do not need a separate package. You only need to configure the engine with an async driver such as asyncpg for PostgreSQL or aiomysql for MySQL.

Configuring the Async Engine

Start by installing the database driver that matches your database. For PostgreSQL, that means asyncpg. Then create an async engine with create_async_engine instead of the synchronous create_engine.

from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/mydb",
    echo=True,
)

The connection string uses the +asyncpg suffix to tell SQLAlchemy which driver to use. The echo=True option prints SQL statements to the console during development. In production you will usually set it to False and rely on a logging configuration instead.

The async engine manages a connection pool just like the sync engine. The pool size and timeout settings are controlled by the same pool_size, max_overflow, and pool_timeout parameters. You can pass them to create_async_engine when your workload requires a specific pool behavior.

Creating an Async Session Factory

A session is the unit of work that tracks changes and talks to the database. With async SQLAlchemy, you create a session factory using async_sessionmaker.

from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession

SessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

expire_on_commit=False is important for async usage. When it is True (the default), SQLAlchemy expires all object attributes after a commit. Accessing those attributes later triggers a lazy load, which is a blocking operation in an async context. Setting it to False avoids that surprise and keeps your objects usable after the transaction ends.

Using Dependency Injection for Database Sessions

FastAPI's dependency injection system is the natural place to manage a session's lifecycle. You define a dependency that yields a session and closes it after the request finishes.

from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

async def get_session():
    async with SessionLocal() as session:
        yield session

@app.get("/items")
async def read_items(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(Item))
    return result.scalars().all()

The async with block ensures the session is closed even if an exception occurs. FastAPI calls the dependency before the route handler and cleans it up after the response is sent. This pattern keeps the session scoped to a single request and avoids the problem of sharing a session across concurrent requests.

Querying with Async SQLAlchemy

Async SQLAlchemy uses the same query API as the sync version, but you must await the execution. For example, a simple select statement is executed with await session.execute().

from sqlalchemy import select

async def get_user_by_id(session: AsyncSession, user_id: int):
    stmt = select(User).where(User.id == user_id)
    result = await session.execute(stmt)
    return result.scalar_one_or_none()

The result object behaves the same as in sync SQLAlchemy. You can use scalars(), scalar_one(), scalars().all(), or scalar_one_or_none() depending on how many rows you expect. The key difference is that the actual database round trip happens only when you await the execution, so the event loop can handle other work while the database is processing the query.

Handling Transactions and Commit/Rollback

In async SQLAlchemy, you manage transactions explicitly with await session.commit() and await session.rollback(). If you do not commit, the transaction remains open and will be rolled back when the session closes.

async def create_item(session: AsyncSession, data: ItemCreate):
    item = Item(**data.model_dump())
    session.add(item)
    await session.commit()
    await session.refresh(item)
    return item

await session.refresh(item) is often needed after a commit because the object may not contain database-generated values such as an auto-incrementing primary key. With expire_on_commit=False, the object is not expired, but it also does not automatically receive the new ID unless you refresh it. The refresh issues a SELECT to fetch the current state from the database.

If an error occurs during the transaction, you should roll back to release the connection and reset the session state. A common pattern is to wrap the operation in a try/except block and roll back on failure.

Common Pitfalls and Runtime Behavior

One of the most frequent mistakes is using a sync session inside an async route. The application may appear to work during light testing, but under concurrent load the event loop blocks and response times degrade. Another pitfall is lazy loading. In async SQLAlchemy, lazy loading is not supported in the same way as in sync. If you access an unloaded relationship after the session is closed, you get an error. You must use eager loading options like selectinload or joinedload when you know you will need related objects.

Connection pool exhaustion is another concern. If you open a session but do not close it, the underlying connection is never returned to the pool. FastAPI's dependency injection with async with prevents that leak, but if you create sessions manually, you must ensure they are closed.

Finally, remember that the async engine does not change the database schema or migration behavior. You still use Alembic for migrations, and you can run them with a sync engine if needed. The async engine is only for runtime database access.