FastAPI + Async SQLAlchemy 2.0: High-Performance Python APIs

FastAPI + SQLAlchemy 2.0 Async

Async sessions + FastAPI's concurrent request handling = maximum throughput.

Database Setup

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

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db", pool_size=20)
SessionLocal = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

class Base(DeclarativeBase): pass

Models (SQLAlchemy 2.0 Style)

from sqlalchemy.orm import Mapped, mapped_column, relationship

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True, index=True)
    is_active: Mapped[bool] = mapped_column(default=True)
    posts: Mapped[list["Post"]] = relationship(back_populates="author", lazy="selectin")

Dependency Injection

async def get_db():
    async with SessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Repository Pattern

from sqlalchemy import select

class UserRepo:
    def __init__(self, db: AsyncSession): self.db = db

    async def get_by_id(self, uid: int):
        r = await self.db.execute(select(User).where(User.id == uid))
        return r.scalar_one_or_none()

    async def create(self, email: str, pw: str):
        u = User(email=email, hashed_password=pw)
        self.db.add(u)
        await self.db.flush()
        await self.db.refresh(u)
        return u

-> Validate schemas with the JSON Viewer.