Home/Learn/FastAPI/CRUD with SQLAlchemy — The Full Resource Pattern

CRUD with SQLAlchemy — The Full Resource Pattern

Intermediate

Databases

The complete CRUD recipe: Pydantic In/Out schemas with from_attributes, add/commit/refresh on create, exclude_unset for PATCH, IntegrityError → 409, and eager loading to kill N+1 queries.

Overview

This chapter assembles everything into the endpoint set you will write a hundred times. Pydantic schemas talk to ORM objects through from_attributes=True, so returning a Student model object just works with response_model. Create is add → commit → refresh (refresh pulls the DB-generated id back). Update reuses the exclude_unset PATCH pattern against a loaded object. Unique violations surface as IntegrityError, which you translate to 409 Conflict rather than letting it 500. The performance trap hiding in every listing endpoint is N+1 — one query for students, then one more per student for offers — invisible in dev with 20 rows, lethal in production with 20,000; selectinload fixes it in one line.

Create & Read — Schemas Bridge ORM and JSON

from_attributes lets response_model serialize ORM objects directly (nested relationships included). Create returns 201 with the fresh object; duplicates become a clean 409 via IntegrityError.

add → commit → refresh; IntegrityError → 409, never 500

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

Update, Delete & the N+1 Trap

PATCH = load, apply exclude_unset fields, commit. DELETE returns 204. And the listing endpoint gets selectinload — check echo=True output once and count your queries.

exclude_unset PATCH, 204 DELETE, selectinload vs 5,001 queries

from sqlalchemy import select
from sqlalchemy.orm import selectinload

class StudentPatch(BaseModel):
    name: str | None = None
    branch: str | None = None
    cgpa: float | None = None

@app.patch("/students/{student_id}", response_model=StudentOut)
def update_student(student_id: int, patch: StudentPatch,
                   db: Session = Depends(get_db)):
    student = db.get(Student, student_id)
    if not student:
        raise HTTPException(404, "student not found")
    for field, value in patch.model_dump(exclude_unset=True).items():
        setattr(student, field, value)      # only fields the client sent
    db.commit()
    db.refresh(student)
    return student

@app.delete("/students/{student_id}", status_code=204)
def delete_student(student_id: int, db: Session = Depends(get_db)):
    student = db.get(Student, student_id)
    if not student:
        raise HTTPException(404, "student not found")
    db.delete(student)
    db.commit()

# ── N+1: the listing-endpoint killer ──
@app.get("/students", response_model=list[StudentOut])
def list_students(db: Session = Depends(get_db)):
    rows = db.execute(
        select(Student).options(selectinload(Student.offers))  # ← the fix
    ).scalars().all()
    return rows
# WITHOUT selectinload: serializing .offers lazy-loads per student —
#   1 query for students + N queries for offers = 5,001 queries for 5,000 rows
# WITH selectinload: 2 queries total (students, then offers WHERE id IN (...))
# Related: joinedload for to-one relations; N+1 is THE ORM interview question.

Key Points to Remember

1ConfigDict(from_attributes=True) lets response_model serialize ORM objects, nesting included
2Create: add → commit → refresh; catch IntegrityError → 409
3PATCH: load object, setattr only model_dump(exclude_unset=True) fields
4N+1 on listings: selectinload (to-many) / joinedload (to-one) — count queries once with echo=True

Interview Questions

What is the N+1 query problem? Show it in a students-with-offers endpoint and fix it.

HardSwiggy

Why db.refresh() after commit on create? What is stale without it?

MediumMeesho

Two requests register the same email simultaneously — what happens at each layer, and what does the loser receive?

HardGoogle

Ask Aria about CRUD with SQLAlchemy — The Full Resource Pattern

Your personal AI tutor — ask anything about this concept

Revision Status

Personal Notes

Discussion

Loading discussion…

SQLAlchemy Setup — Engine, Sessions & Models

Alembic Migrations — Evolving the Schema Safely