SQLAlchemy 2.0 Reference: Mapped, select(), Relationships, Async, Alembic

from sqlalchemy import create_engine, String, Integer, ForeignKey, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from datetime import datetime

# 2.0 style: DeclarativeBase + Mapped annotations:
class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id:         Mapped[int] = mapped_column(primary_key=True)
    name:       Mapped[str] = mapped_column(String(100), nullable=False)
    email:      Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    # Mapped[str | None] = Optional field (nullable):
    avatar_url: Mapped[str | None] = mapped_column(String(500))
    # Relationship:
    posts: Mapped[list["Post"]] = relationship("Post", back_populates="author", lazy="select")

class Post(Base):
    __tablename__ = "posts"
    id:        Mapped[int] = mapped_column(primary_key=True)
    title:     Mapped[str] = mapped_column(String(500))
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
    author:    Mapped["User"] = relationship("User", back_populates="posts")

# Sync engine:
engine = create_engine("postgresql+psycopg2://user:pass@localhost/db",
    pool_size=10, max_overflow=20, pool_pre_ping=True)
SessionLocal = sessionmaker(engine, expire_on_commit=False)

# Async engine (with asyncpg):
async_engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db",
    pool_size=10, max_overflow=20)
AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

# Create tables:
Base.metadata.create_all(engine)

from sqlalchemy import select, and_, or_, desc, func

with Session(engine) as session:
    # Basic select (2.0 style — NOT session.query()):
    stmt = select(User).where(User.email == "alice@example.com")
    user = session.scalars(stmt).first()   # scalars() unwraps the Row

    # Multiple conditions:
    stmt = select(User).where(
        and_(User.name.like("A%"), User.created_at >= "2026-01-01")
    )
    users = session.scalars(stmt).all()

    # OR:
    stmt = select(User).where(or_(User.name == "Alice", User.name == "Bob"))

    # IN:
    stmt = select(User).where(User.id.in_([1, 2, 3]))

    # ORDER BY + LIMIT + OFFSET:
    stmt = select(User).order_by(desc(User.created_at)).limit(20).offset(40)

    # COUNT:
    count = session.scalar(select(func.count()).select_from(User))

    # JOIN (explicit):
    stmt = (select(User, Post)
        .join(Post, Post.author_id == User.id)
        .where(Post.published == True)
        .order_by(desc(Post.created_at)))
    rows = session.execute(stmt).all()
    for user, post in rows:
        print(user.name, post.title)

    # EXISTS:
    from sqlalchemy import exists
    stmt = select(User).where(exists(select(Post).where(Post.author_id == User.id)))

from sqlalchemy.orm import selectinload, joinedload, load_only

# lazy="select" (default): relationship loaded on attribute access — causes N+1!
# Fix: use selectinload or joinedload at query time:

with Session(engine) as session:
    # selectinload: one extra SELECT for the relationship (best for collections):
    stmt = select(User).options(selectinload(User.posts))
    users = session.scalars(stmt).all()
    for user in users:
        print(user.posts)   # no extra query! loaded in one SELECT ... IN (...)

    # joinedload: LEFT OUTER JOIN (best for single objects, one-to-one/many-to-one):
    stmt = select(Post).options(joinedload(Post.author))
    posts = session.scalars(stmt).unique().all()   # .unique() needed with joinedload

    # Nested loading:
    stmt = select(User).options(
        selectinload(User.posts).selectinload(Post.comments)
    )

    # load_only: select specific columns only (avoid loading large fields):
    stmt = select(User).options(load_only(User.id, User.name, User.email))

    # N+1 detection: enable query logging:
    # create_engine(..., echo=True)
    # or: logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)

# expire_on_commit=False: keep attribute values after commit (important for async):
# Without it, accessing user.name after session.commit() triggers a new SELECT

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from fastapi import Depends

async_engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

# FastAPI dependency:
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

# Route using async session:
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    stmt = select(User).where(User.id == user_id)
    user = await db.scalar(stmt)
    if not user:
        raise HTTPException(status_code=404)
    return user

@app.post("/users")
async def create_user(data: UserCreate, db: AsyncSession = Depends(get_db)):
    user = User(name=data.name, email=data.email)
    db.add(user)
    await db.commit()
    await db.refresh(user)   # reload from DB (gets server defaults like created_at)
    return user

# Async relationship loading (lazy loading doesn't work in async!):
stmt = select(User).options(selectinload(User.posts)).where(User.id == user_id)
user = await db.scalar(stmt)
# user.posts is populated — no lazy load needed

# pip install alembic

# Initialize:
alembic init alembic

# alembic/env.py — point at your models:
from myapp.models import Base
target_metadata = Base.metadata

# alembic.ini:
# sqlalchemy.url = postgresql+psycopg2://user:pass@localhost/db
# Or use env var:
# sqlalchemy.url = %(DATABASE_URL)s

# Generate migration from model changes (autogenerate):
alembic revision --autogenerate -m "add avatar_url to users"

# Review the generated file in alembic/versions/ before applying!

# Apply all pending migrations:
alembic upgrade head

# Downgrade one step:
alembic downgrade -1

# Show current version:
alembic current

# Show pending:
alembic history --verbose

# Alembic does NOT detect: renamed columns, renamed tables, constraint changes on some DBs
# Always review autogenerated migrations — they can drop+recreate instead of alter

# Multi-env (staging vs prod) via config:
alembic -x env=prod upgrade head
# In env.py: config.get_main_option("env") or os.getenv("ENV")