SQLAlchemy ORM Setup for Python Web Application

SQLAlchemy Setup for Python Web Application

SQLAlchemy is the de facto standard for working with relational databases in Python. In version 2.0, the query syntax changed significantly: deprecated session.query(Model) replaced with select(Model). Setup from scratch assumes version 2.x.

pip install sqlalchemy[asyncio] asyncpg alembic
# or for synchronous:
pip install sqlalchemy psycopg2-binary alembic

Engine and Session

For FastAPI and async frameworks, use async engine:

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

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

engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
    echo=False,
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

class Base(DeclarativeBase):
    pass

Dependency for FastAPI

# app/deps.py
from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import AsyncSessionLocal

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Models

# app/models/user.py
from datetime import datetime
from sqlalchemy import String, Enum, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
import enum

class UserRole(enum.Enum):
    admin = "admin"
    editor = "editor"
    viewer = "viewer"

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(320), unique=True, nullable=False)
    password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
    role: Mapped[UserRole] = mapped_column(
        Enum(UserRole), default=UserRole.viewer, nullable=False
    )
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), nullable=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), onupdate=func.now(), nullable=False
    )

    posts: Mapped[list["Post"]] = relationship(
        back_populates="author", lazy="selectin"
    )

2.0 Style Queries

from sqlalchemy import select
from app.models.user import User
from app.models.post import Post

async def get_published_posts(
    db: AsyncSession,
    limit: int = 20,
    offset: int = 0,
) -> list[Post]:
    stmt = (
        select(Post)
        .join(Post.author)
        .where(Post.status == "published")
        .order_by(Post.created_at.desc())
        .limit(limit)
        .offset(offset)
    )
    result = await db.execute(stmt)
    return list(result.scalars().all())

async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
    stmt = select(User).where(User.email == email)
    result = await db.execute(stmt)
    return result.scalar_one_or_none()

Transactions

async def transfer_ownership(
    db: AsyncSession,
    post_id: int,
    new_author_id: int,
) -> None:
    async with db.begin_nested():
        post = await db.get(Post, post_id)
        if post is None:
            raise ValueError(f"Post {post_id} not found")
        post.author_id = new_author_id

Alembic Migrations

alembic init -t async alembic

Edit alembic/env.py for async:

from logging.config import fileConfig
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context
from app.database import Base
import app.models

config = context.config
target_metadata = Base.metadata

def run_migrations_online():
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
    )

    async def do_run():
        async with connectable.connect() as connection:
            await connection.run_sync(
                context.configure,
                connection=connection,
                target_metadata=target_metadata,
                compare_type=True,
            )
            async with context.begin_transaction():
                await connection.run_sync(context.run_migrations)

    import asyncio
    asyncio.run(do_run())

run_migrations_online()

Timeline

Initial setup for FastAPI project: 1 day. Migrating existing project from 1.4 (legacy Query API) to 2.0: 1–3 days depending on models volume.