Async SQLAlchemy Integration with FastAPI

Why Async SQLAlchemy?

FastAPI is async by design. If your database queries run synchronously, they block the event loop and negate FastAPI’s concurrency advantages. Async SQLAlchemy lets your app handle other requests while waiting for database responses — the same way await works with HTTP calls.

SQLAlchemy 2.0 introduced native async support. Combined with asyncpg (for PostgreSQL) or aiosqlite (for SQLite), you get non-blocking database access that matches FastAPI’s architecture.

Project Setup

pip install fastapi sqlalchemy[asyncio] asyncpg alembic
# For development/testing with SQLite:
pip install aiosqlite

Project Structure

app/
    __init__.py
    main.py
    database.py
    models.py
    schemas.py
    crud.py
    routers/
        items.py

Database Configuration

# app/database.py
from sqlalchemy.ext.asyncio import (
    create_async_engine,
    async_sessionmaker,
    AsyncSession,
)
from sqlalchemy.orm import DeclarativeBase

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

# SQLite with aiosqlite (for development)
# DATABASE_URL = "sqlite+aiosqlite:///./dev.db"

engine = create_async_engine(
    DATABASE_URL,
    echo=True,         # Log SQL queries (disable in production)
    pool_size=20,       # Connection pool size
    max_overflow=10,    # Extra connections beyond pool_size
    pool_timeout=30,    # Seconds to wait for a connection
    pool_recycle=1800,  # Recycle connections after 30 minutes
)

async_session = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,  # Keep attributes accessible after commit
)

class Base(DeclarativeBase):
    pass

async def get_db() -> AsyncSession:
    """Dependency that provides a database session per request."""
    async with async_session() as session:
        try:
            yield session
        finally:
            await session.close()

Setting expire_on_commit=False is important for async usage. Without it, accessing model attributes after a commit triggers a lazy load, which fails in async context because lazy loading is synchronous.

Defining Models

# app/models.py
from sqlalchemy import String, Integer, Float, ForeignKey, DateTime, Boolean
from sqlalchemy.orm import Mapped, mapped_column, relationship
from datetime import datetime
from typing import List, Optional
from app.database import Base

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(100), unique=True)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, default=datetime.utcnow
    )

    # Relationship
    items: Mapped[List["Item"]] = relationship(
        back_populates="owner", lazy="selectin"
    )

class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(100), index=True)
    description: Mapped[Optional[str]] = mapped_column(String(500))
    price: Mapped[float] = mapped_column(Float)
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(
        DateTime, default=datetime.utcnow
    )

    # Relationship
    owner: Mapped["User"] = relationship(back_populates="items")

Using lazy="selectin" on relationships is critical for async. The default lazy="select" triggers synchronous lazy loading, which raises errors in async code. selectin loads related objects in a separate SELECT using IN, and it works with async sessions.

Pydantic Schemas

# app/schemas.py
from pydantic import BaseModel, ConfigDict
from datetime import datetime
from typing import Optional, List

class ItemCreate(BaseModel):
    name: str
    description: Optional[str] = None
    price: float

class ItemResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    name: str
    description: Optional[str]
    price: float
    owner_id: int
    created_at: datetime

class UserCreate(BaseModel):
    username: str
    email: str

class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    username: str
    email: str
    is_active: bool
    created_at: datetime
    items: List[ItemResponse] = []

ConfigDict(from_attributes=True) (formerly orm_mode) lets Pydantic read data from SQLAlchemy model attributes.

CRUD Operations

# app/crud.py
from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.models import User, Item
from app.schemas import UserCreate, ItemCreate
from typing import Optional, List

async def create_user(db: AsyncSession, user_data: UserCreate) -> User:
    user = User(**user_data.model_dump())
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

async def get_user(db: AsyncSession, user_id: int) -> Optional[User]:
    result = await db.execute(
        select(User)
        .options(selectinload(User.items))
        .where(User.id == user_id)
    )
    return result.scalar_one_or_none()

async def get_users(
    db: AsyncSession, skip: int = 0, limit: int = 100
) -> List[User]:
    result = await db.execute(
        select(User)
        .options(selectinload(User.items))
        .offset(skip)
        .limit(limit)
    )
    return list(result.scalars().all())

async def create_item(
    db: AsyncSession, item_data: ItemCreate, owner_id: int
) -> Item:
    item = Item(**item_data.model_dump(), owner_id=owner_id)
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return item

async def update_item(
    db: AsyncSession, item_id: int, updates: dict
) -> Optional[Item]:
    await db.execute(
        update(Item)
        .where(Item.id == item_id)
        .values(**updates)
    )
    await db.commit()
    return await get_item(db, item_id)

async def delete_item(db: AsyncSession, item_id: int) -> bool:
    result = await db.execute(
        delete(Item).where(Item.id == item_id)
    )
    await db.commit()
    return result.rowcount > 0

async def get_item(db: AsyncSession, item_id: int) -> Optional[Item]:
    result = await db.execute(
        select(Item).where(Item.id == item_id)
    )
    return result.scalar_one_or_none()

Every query uses await. The select() construct builds the query, and db.execute() sends it to the database asynchronously.

FastAPI Router

# app/routers/items.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app import crud, schemas

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/", response_model=schemas.UserResponse, status_code=201)
async def create_user(
    user: schemas.UserCreate,
    db: AsyncSession = Depends(get_db)
):
    existing = await crud.get_user_by_username(db, user.username)
    if existing:
        raise HTTPException(409, "Username already taken")
    return await crud.create_user(db, user)

@router.get("/{user_id}", response_model=schemas.UserResponse)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await crud.get_user(db, user_id)
    if not user:
        raise HTTPException(404, "User not found")
    return user

@router.post(
    "/{user_id}/items",
    response_model=schemas.ItemResponse,
    status_code=201
)
async def create_item_for_user(
    user_id: int,
    item: schemas.ItemCreate,
    db: AsyncSession = Depends(get_db)
):
    user = await crud.get_user(db, user_id)
    if not user:
        raise HTTPException(404, "User not found")
    return await crud.create_item(db, item, owner_id=user_id)

App Entrypoint with Table Creation

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.database import engine, Base
from app.routers import items

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Create tables on startup (use Alembic in production)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Cleanup
    await engine.dispose()

app = FastAPI(lifespan=lifespan)
app.include_router(items.router)

Transactions

For operations that must succeed or fail together, use explicit transactions.

async def transfer_item(
    db: AsyncSession,
    item_id: int,
    from_user_id: int,
    to_user_id: int
):
    async with db.begin():
        # Both operations happen in a single transaction
        item = await get_item(db, item_id)
        if not item or item.owner_id != from_user_id:
            raise ValueError("Item not found or not owned by sender")

        to_user = await crud.get_user(db, to_user_id)
        if not to_user:
            raise ValueError("Recipient not found")

        item.owner_id = to_user_id
        # Transaction commits when the context manager exits
        # Rolls back if any exception is raised

Alembic Migrations with Async

Initialize Alembic and configure it for async.

alembic init alembic

# alembic/env.py (key changes)
from app.database import Base, DATABASE_URL
from sqlalchemy.ext.asyncio import create_async_engine

config.set_main_option("sqlalchemy.url", DATABASE_URL)
target_metadata = Base.metadata

async def run_async_migrations():
    engine = create_async_engine(DATABASE_URL)
    async with engine.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await engine.dispose()

def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()

# Generate a migration
alembic revision --autogenerate -m "create users and items"

# Apply migrations
alembic upgrade head

Key Takeaways

Async SQLAlchemy 2.0 integrates naturally with FastAPI’s async architecture. Use asyncpg for PostgreSQL, set expire_on_commit=False on sessions, and use lazy="selectin" or explicit selectinload() for relationships to avoid synchronous lazy loading errors. Structure your code with separate database config, models, schemas, and CRUD modules. Use Alembic for migrations and explicit transactions for multi-step operations. The session-per-request pattern through FastAPI’s Depends() keeps your database connections clean and isolated.