Async SQLAlchemy 2.0 with FastAPI and PostgreSQL

SQLAlchemy 2.0 brought a cleaner, fully typed API and first-class async support. Combined with the asyncpg driver it's a solid choice for FastAPI apps.

Install

pip install "sqlalchemy[asyncio]" asyncpg alembic

Engine and session

# db.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase

engine = create_async_engine("postgresql+asyncpg://app:secret@localhost/app", pool_size=10)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

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

Typed models

from datetime import datetime
from sqlalchemy import ForeignKey, String, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

class Author(Base):
    __tablename__ = "authors"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(120))
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    published_at: Mapped[datetime | None]
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))
    author: Mapped[Author] = relationship(back_populates="posts")

Queries

from sqlalchemy import select
from sqlalchemy.orm import selectinload

@app.get("/posts")
async def list_posts(db: DbSession):
    stmt = (
        select(Post)
        .where(Post.published_at.is_not(None))
        .options(selectinload(Post.author))      # avoid lazy loading in async code
        .order_by(Post.published_at.desc())
        .limit(20)
    )
    return (await db.scalars(stmt)).all()

@app.post("/posts", status_code=201)
async def create_post(data: PostIn, db: DbSession):
    post = Post(**data.model_dump())
    db.add(post)
    await db.commit()
    await db.refresh(post)
    return post

Gotchas

• Lazy-loading relationships doesn't work in async code; load them explicitly with selectinload or joinedload.
• expire_on_commit=False lets you return objects after committing.
• Manage schema changes with Alembic migrations (alembic init -t async migrations), not create_all, in real projects.