Async SQLAlchemy Engine in FastAPI - The Guide - Software Engineering with MJ

If you’ve been building FastAPI applications, you probably already appreciate its async-first nature — making it a great choice for high-performance APIs.
But if your database layer isn’t async-friendly, you’re still bottlenecked by synchronous I/O. That’s where Async SQLAlchemy comes in.

In this guide, we’ll cover:

Why use Async SQLAlchemy with FastAPI
Core concepts of SQLAlchemy’s async API
How to set up an async engine & session
Writing async database queries
Handling transactions
Common pitfalls & best practices
Full working example

Why Go Async with SQLAlchemy in FastAPI?

By default, SQLAlchemy is blocking. In an async web app, a blocking database call means your event loop waits — preventing other requests from being handled until the query completes.

With Async SQLAlchemy:

Queries don’t block the event loop
Multiple requests can be served concurrently
Better scalability under load
Works with async DB drivers like asyncpg (PostgreSQL), aiosqlite (SQLite), or aiomysql (MySQL)

How Async SQLAlchemy Works

SQLAlchemy introduced the async engine in version 1.4 and improved it in 2.0.

Key changes from sync to async:

Use create_async_engine() instead of create_engine()
Sessions are AsyncSession
All DB calls must be awaited
DB drivers must support async (e.g., asyncpg for PostgreSQL)

Installing Requirements

For PostgreSQL:

pip install fastapi[all] sqlalchemy[asyncio] asyncpg

For SQLite (async testing):

pip install fastapi[all] sqlalchemy[asyncio] aiosqlite

Setting up Async SQLAlchemy with FastAPI

Create the database engine:

Create a file named database.py and put these content in it:

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

DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/mydb"

engine = create_async_engine(
    DATABASE_URL,
    echo=True,
    future=True, # Use future=True for SQLAlchemy 2.0 style
)

AsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

Base = declarative_base()

Dependency for Async Session in FastAPI:

Create a file named “deps.py” and put these contents in it:

from app.database import AsyncSessionLocal
from sqlalchemy.ext.asyncio import AsyncSession


async def get_session() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

Create a Model:

Create a file named “models.py” besides “database.py” and “deps.py“:

from sqlalchemy import Column, Integer, String

from app.database import Base


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

Initialize Database Tables:

Create a file named “init_db.py“:

import asyncio

from app.database import Base, engine
from app.models import * # noqa: F401, F403

async def init_models():
    async with engine.begin() as conn:
        # Create all tables in the database
        await conn.run_sync(Base.metadata.create_all)


if __name__ == "__main__":
    asyncio.run(init_models())

Writing Async Queries in FastAPI:

Create a file named “main.py“:

from fastapi import Depends, FastAPI
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.exc import IntegrityError
from fastapi import HTTPException
from app.deps import get_session
from app.models import User, UserCreate, UserResponse

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    from app.init_db import init_models
    await init_models()
    yield


app = FastAPI(lifespan=lifespan)


@app.post("/users/", response_model=UserResponse)
async def create_user(
    user: UserCreate, session: AsyncSession = Depends(get_session)
):
    try:
        async with session.begin():  # Auto commit/rollback
            new_user = User(**user.model_dump())
            session.add(new_user)

        # refresh after commit if you need DB-generated values (id, timestamps, etc.)
        await session.refresh(new_user)

        return new_user

    except IntegrityError:
        raise HTTPException(
            status_code=400,
            detail="User with provided unique fields already exists."
        )


@app.get("/users/", response_model=list[UserResponse])
async def list_users(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User))
    users = result.scalars().all()
    return users

Postgres Setup:

Create a docker compose file for PostgreSQL:

services:
  db:
    image: postgres:latest
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
      POSTGRES_DB: mydb
    ports:
      - "5432:5432"
    volumes:
      - db_data:/var/lib/postgresql/data

volumes:
  db_data:

Project Structure:

Your project’s structure should look like this:

Running the Project

To run the project, execute these commands in order:

docker compose up –build -d
uvicorn app.main:app –host=0.0.0.0

Testing the Project

First create a user using this command:

curl --location 'http://localhost:8000/users/' \
--header 'Content-Type: application/json' \
--data-raw '{
    "name": "Test User 1",
    "email": "[email protected]"
}'

Then get all users using this curl command:

curl --location 'http://localhost:8000/users/'

Common Pitfalls

Mixing Sync and Async Engines
Don’t call create_engine() in async apps — it will block.
Using Non-Async Drivers
Always use async-supported drivers like asyncpg, aiosqlite, or aiomysql.
Forgetting await
Any ORM query or commit must be awaited.
Not Managing Sessions Properly
Always close sessions with async with or try/finally.

Best Practices

One session per request – never reuse across requests.
Use expire_on_commit=False for better async behavior.
Keep queries short — async helps with concurrency, not query speed.
Use Pydantic models for request/response validation.
For large datasets, use stream() or pagination.

Performance Boost

Switching from sync to async SQLAlchemy can yield significant improvements under load.
In testing with locust, async DB queries in FastAPI handled 3–5x more requests/sec for the same hardware compared to sync queries.

Final Thoughts

Async SQLAlchemy in FastAPI isn’t magic — your queries still take time — but it eliminates unnecessary blocking, making your API far more scalable.

If you’re already running FastAPI and using SQLAlchemy, the migration is straightforward: swap to create_async_engine, use async sessions, and ensure your DB driver supports async.

Async SQLAlchemy Engine in FastAPI – The Guide