Databases — Cheat Sheet

SQLAlchemy Setup — Engine, Sessions & Models

# pip install sqlalchemy psycopg2-binary

# ── app/core/db.py ──────────────────────────
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

engine = create_engine(
    "postgresql://app:app@localhost:5432/placement",   # from Settings in real code
    pool_size=5,           # steady connections held open
    max_overflow=10,       # extra under burst (returned when idle)
    pool_pre_ping=True,    # test connection before use — survives DB restarts
    echo=False,            # True in dev = log every SQL statement
)
SessionLocal = sessionmaker(bind=engine, autoflush=False)

class Base(DeclarativeBase):
    pass

# ── app/models/student.py ───────────────────
from sqlalchemy import String, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship

class Student(Base):
    __tablename__ = "students"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(60))
    email: Mapped[str] = mapped_column(String(120), unique=True, index=True)
    cgpa: Mapped[float] = mapped_column(default=0.0)
    branch: Mapped[str] = mapped_column(String(10), index=True)
    offers: Mapped[list["Offer"]] = relationship(back_populates="student")

class Offer(Base):
    __tablename__ = "offers"
    id: Mapped[int] = mapped_column(primary_key=True)
    student_id: Mapped[int] = mapped_column(ForeignKey("students.id"), index=True)
    company: Mapped[str] = mapped_column(String(80))
    ctc_lpa: Mapped[float]
    student: Mapped[Student] = relationship(back_populates="offers")

# Dev bootstrap only — real schema changes go through Alembic:
# Base.metadata.create_all(engine)

CRUD with SQLAlchemy — The Full Resource Pattern

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, ConfigDict, EmailStr
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

app = FastAPI()

class StudentIn(BaseModel):
    name: str
    email: EmailStr
    branch: str
    cgpa: float = 0.0

class OfferOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)   # read from ORM attributes
    company: str
    ctc_lpa: float

class StudentOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str
    branch: str
    cgpa: float
    offers: list[OfferOut] = []                       # nested ORM relationship

@app.post("/students", response_model=StudentOut, status_code=201)
def create_student(payload: StudentIn, db: Session = Depends(get_db)):
    student = Student(**payload.model_dump())
    db.add(student)
    try:
        db.commit()                    # INSERT happens here
    except IntegrityError:             # unique email hit
        db.rollback()
        raise HTTPException(409, f"email {payload.email} already registered")
    db.refresh(student)                # pull DB-generated id (and defaults)
    return student                     # ORM object → StudentOut, automatically

@app.get("/students/{student_id}", response_model=StudentOut)
def read_student(student_id: int, db: Session = Depends(get_db)):
    student = db.get(Student, student_id)
    if not student:
        raise HTTPException(404, "student not found")
    return student

Alembic Migrations — Evolving the Schema Safely

pip install alembic
alembic init alembic                     # creates alembic/ + alembic.ini

# ── alembic/env.py — the two lines that matter ──
from app.core.db import Base
from app.models import student, offer    # import ALL model modules (registers tables)
target_metadata = Base.metadata
# and set the URL from settings, not alembic.ini:
# config.set_main_option("sqlalchemy.url", get_settings().database_url)

# ── the daily loop ──
# 1. Edit the model: add  backlogs: Mapped[int] = mapped_column(default=0)
# 2. Generate:
alembic revision --autogenerate -m "add backlogs to students"
# 3. REVIEW alembic/versions/9f2c_add_backlogs_to_students.py:
def upgrade():
    op.add_column("students",
        sa.Column("backlogs", sa.Integer(), nullable=False,
                  server_default="0"))   # ← you often ADD this by hand:
                                         #    NOT NULL on a full table needs a default
def downgrade():
    op.drop_column("students", "backlogs")
# 4. Apply:
alembic upgrade head

# Everyday commands:
alembic current            # what version is this DB on?
alembic history            # the chain of migrations
alembic downgrade -1       # step back one (dev only, usually)

Async SQLAlchemy — asyncpg and AsyncSession

# pip install sqlalchemy[asyncio] asyncpg

# ── app/core/db.py ──────────────────────────
from sqlalchemy.ext.asyncio import (
    AsyncSession, async_sessionmaker, create_async_engine,
)

engine = create_async_engine(
    "postgresql+asyncpg://app:app@localhost:5432/placement",  # ← asyncpg driver
    pool_size=5, max_overflow=10, pool_pre_ping=True,
)
AsyncSessionLocal = async_sessionmaker(bind=engine, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as db:     # async context manager closes it
        try:
            yield db
            await db.commit()
        except Exception:
            await db.rollback()
            raise

# ── endpoints: await every DB touch ─────────
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import select

app = FastAPI()

@app.get("/students/{student_id}")
async def get_student(student_id: int, db: AsyncSession = Depends(get_db)):
    student = await db.get(Student, student_id)
    if not student:
        raise HTTPException(404, "student not found")
    return {"id": student.id, "name": student.name}

@app.get("/students")
async def search(branch: str, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(Student).where(Student.branch == branch).limit(20)
    )
    return [{"id": s.id, "name": s.name} for s in result.scalars()]
# While these awaits wait on Postgres, the SAME worker serves other requests.

Pagination, Filtering & Sorting — List Endpoints Done Right

from fastapi import Depends, FastAPI, HTTPException, Query
from pydantic import BaseModel
from sqlalchemy import func, select
from sqlalchemy.orm import Session

app = FastAPI()

SORTABLE = {"cgpa": Student.cgpa, "name": Student.name, "id": Student.id}

class Page(BaseModel):
    items: list[StudentOut]
    total: int
    page: int
    size: int

@app.get("/students", response_model=Page)
def list_students(
    db: Session = Depends(get_db),
    branch: str | None = None,
    min_cgpa: float = Query(default=0, ge=0, le=10),
    q: str | None = Query(default=None, min_length=2),     # name search
    sort: str = Query(default="id"),
    order: str = Query(default="asc", pattern="^(asc|desc)$"),
    page: int = Query(default=1, ge=1),
    size: int = Query(default=20, ge=1, le=100),           # hard cap, always
):
    if sort not in SORTABLE:                # whitelist — NOT getattr(Student, sort)
        raise HTTPException(422, f"sort must be one of {sorted(SORTABLE)}")

    filters = [Student.cgpa >= min_cgpa]
    if branch:
        filters.append(Student.branch == branch)
    if q:
        filters.append(Student.name.ilike(f"%{q}%"))       # parameterised — safe

    col = SORTABLE[sort]
    stmt = (select(Student).where(*filters)
            .order_by(col.desc() if order == "desc" else col.asc())
            .offset((page - 1) * size).limit(size))

    total = db.execute(select(func.count()).select_from(Student).where(*filters)).scalar()
    items = db.execute(stmt).scalars().all()
    return Page(items=items, total=total, page=page, size=size)

# GET /students?branch=CS&min_cgpa=8&sort=cgpa&order=desc&page=2&size=25
# → {"items":[...25...], "total":312, "page":2, "size":25}