Rewrite every route to use PostgreSQL through SQLAlchemy, type every response with response_model, and organise the code into a router and a CRUD layer.
Where StudyDesk is now
src/studydesk/
├── __init__.py ← lifespan, CORS, DbSession, GET /db-check; CRUD still on JSON
├── config.py
├── database.py ← engine, SessionLocal, Base, get_db
├── models.py ← SQLAlchemy Student table
├── schemas.py ← StudentCreate, StudentUpdate, StudentOut, StudentListOut
├── seed.py
├── utils.py ← JSON helpers (to be deleted)
└── data/students.json
By the end of this article, all five student routes read and write PostgreSQL, every response has a declared response_model, the routes live in routers/students.py, and the database logic lives in crud.py. utils.py is gone.
Part 1: Same API, New Engine Room
Aman has a database with ten students and five routes that still read a text file. Today they switch.
Neha sets two rules before he starts:
- The API's shape does not change. Same paths, same query parameters, same request bodies, same response JSON, same status codes. Karan's frontend must keep working without a single edit.
- Routes do not write SQL. A route's job is HTTP: read parameters, call something, return a response. The database work goes into separate functions that know nothing about HTTP. That separation is what makes the code testable in Article 13.
Part 2: Organising the Code
Before the routes, two structural changes.
DbSession moves to database.py
The alias was in __init__.py. The router file will need it too, and routers must not import from __init__.py (that would be a circular import). So it moves next to get_db().
src/studydesk/database.py
from collections.abc import Generator
from typing import Annotated
from fastapi import Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
from .config import DATABASE_URL
engine = create_engine(DATABASE_URL, echo=False)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
class Base(DeclarativeBase):
"""Parent class for every SQLAlchemy model."""
def get_db() -> Generator[Session, None, None]:
"""Open a database session for one request, and always close it."""
db = SessionLocal()
try:
yield db
finally:
db.close()
DbSession = Annotated[Session, Depends(get_db)]
A routers package
Right now every route sits in __init__.py. With auth, courses, and enrollments coming, that file would become enormous. FastAPI's answer is APIRouter: a mini-app that holds a group of routes. You define routes on the router exactly as you did on app, then plug the router into the app with one line.
Create the folder and an empty init file:
src/studydesk/routers/
├── __init__.py ← empty file
└── students.py ← the student routes (we write it below)
src/studydesk/routers/__init__.py
(Yes, completely empty. It only marks the folder as a package.)
A crud.py module
CRUD functions: plain Python functions that take a session and some arguments, talk to the database, and return rows or None. No HTTPException, no status codes. The router decides what an empty result means in HTTP terms.
We build crud.py and routers/students.py together, one operation at a time.
Part 3: Fetching a Single Student, with response_model
The CRUD function
src/studydesk/crud.py
from sqlalchemy.orm import Session
from .models import Student
def get_student(db: Session, student_id: int) -> Student | None:
"""Return one student by primary key, or None."""
return db.get(Student, student_id)
db.get(Student, student_id) The simplest query in SQLAlchemy: fetch one row by primary key. It returns a Student object or None. Behind the scenes it runs SELECT * FROM students WHERE id = 3.
The route
src/studydesk/routers/students.py
from typing import Annotated
from fastapi import APIRouter, HTTPException, Path, status
from .. import crud
from ..database import DbSession
from ..schemas import StudentOut
router = APIRouter(prefix="/students", tags=["Students"])
@router.get("/{student_id}", response_model=StudentOut, summary="Get one student by ID")
def get_student(
student_id: Annotated[
int,
Path(
title="Student ID",
description="The unique ID of the student. Must be a positive integer.",
gt=0,
examples=[1],
),
],
db: DbSession,
):
student = crud.get_student(db, student_id)
if student is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Student with id {student_id} not found",
)
return student
Reading it
from .. import crud Two dots: go up one package level. routers/students.py → studydesk/ → find crud.py. Same for ..database and ..schemas.
router = APIRouter(prefix="/students", tags=["Students"]) Every route on this router automatically gets /students in front of its path and the Students tag in /docs. So @router.get("/{student_id}") becomes GET /students/{student_id}. The tag no longer needs repeating on each route.
db: DbSession Dependency injection from Article 7. FastAPI opens a session, injects it, and closes it afterwards.
crud.get_student(db, student_id) The route passes the session along. It does not know or care how the lookup works.
return student A SQLAlchemy row object goes straight back. FastAPI converts it to JSON using the response_model.
What response_model does
@router.get("/{student_id}", response_model=StudentOut, ...)
response_model tells FastAPI: "whatever this function returns, pass it through StudentOut before sending it." That does four things:
- Converts the SQLAlchemy row into
StudentOutusingfrom_attributes=Truefrom Article 7. Without that config, this step would fail. - Adds computed fields.
gradeis calculated frommarksand appears in the JSON, even though the row has nogradeattribute. - Filters. Only fields declared in
StudentOutare sent. IfStudentlater gets a column the client must not see (a password hash in Article 10, for example), it is automatically dropped. The response schema is a whitelist. - Documents.
/docsnow shows the exact response shape for this route, with every field and example. Before, it only said "returns something".
Without response_model, FastAPI tries to serialise the raw row and fails, because a SQLAlchemy object is not JSON-serialisable. With it, the row becomes the same JSON the file version produced.
Plug the router into the app
Replace __init__.py. The JSON routes come out; utils and StudentOut imports come out; the router goes in.
src/studydesk/__init__.py
from contextlib import asynccontextmanager
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import func, select
from .config import ALLOWED_ORIGINS
from .database import Base, DbSession, engine
from .models import Student
from .routers import students
@asynccontextmanager
async def lifespan(app: FastAPI):
Base.metadata.create_all(bind=engine)
yield
app = FastAPI(
title="StudyDesk API",
description="Student records API for a coaching institute.",
version="0.2.0",
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=ALLOWED_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(students.router)
@app.get("/", tags=["General"])
def home():
return {"message": "Welcome to StudyDesk API"}
@app.get("/db-check", tags=["General"], summary="Check the database connection")
def db_check(db: DbSession):
total = db.scalar(select(func.count()).select_from(Student))
return {"database": "ok", "students_in_db": total}
def main() -> None:
"""Start the development server."""
uvicorn.run("studydesk:app", host="127.0.0.1", port=8000, reload=True)
from .routers import students then app.include_router(students.router) The one line that mounts every route in the router onto the app.
Delete utils.py
Nothing imports it anymore. Remove the file. Keep data/students.json only because seed.py reads it.
Test
uv run studydesk
GET /students/3 → 200, Sneha, with "grade": "A". The JSON is identical to Article 5's. GET /students/999 → 404. GET /students/abc → 422. Only the engine room changed.
Part 4: Creating a New Student
The CRUD functions
Two additions: an email lookup (for the 409 check) and the create itself.
src/studydesk/crud.py
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from .models import Student
from .schemas import StudentCreate
def get_student(db: Session, student_id: int) -> Student | None:
"""Return one student by primary key, or None."""
return db.get(Student, student_id)
def get_student_by_email(db: Session, email: str) -> Student | None:
"""Return the student with this email (case-insensitive), or None."""
stmt = select(Student).where(func.lower(Student.email) == email.lower())
return db.scalars(stmt).first()
def create_student(db: Session, payload: StudentCreate) -> Student:
"""Insert a new student and return the saved row (with its id)."""
student = Student(**payload.model_dump())
db.add(student)
db.commit()
db.refresh(student)
return student
Reading it
select(Student).where(...) The SQLAlchemy 2.0 way to build a query. select(Student) means "rows from the students table". .where(...) adds a condition. func.lower(Student.email) applies SQL's LOWER() to the column, so the comparison is case-insensitive in the database, not in Python.
db.scalars(stmt).first() Run the statement and return the first row, or None. (.all() would return a list; we will use that for the listing.)
Student(**payload.model_dump()) Build a row object from the validated schema, the same ** unpacking as before. No id: the database assigns it.
db.add(student) → db.commit() Put the row in the session basket, then check out. The INSERT runs on commit.
db.refresh(student) After commit, ask the database for the row again so the object now has its id (and any other database-generated values). Without this, student.id might be None.
The route
src/studydesk/routers/students.py
from typing import Annotated
from fastapi import APIRouter, HTTPException, Path, status
from sqlalchemy.exc import IntegrityError
from .. import crud
from ..database import DbSession
from ..schemas import StudentCreate, StudentOut
router = APIRouter(prefix="/students", tags=["Students"])
@router.get("/{student_id}", response_model=StudentOut, summary="Get one student by ID")
def get_student(
student_id: Annotated[
int,
Path(
title="Student ID",
description="The unique ID of the student. Must be a positive integer.",
gt=0,
examples=[1],
),
],
db: DbSession,
):
student = crud.get_student(db, student_id)
if student is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Student with id {student_id} not found",
)
return student
@router.post(
"",
response_model=StudentOut,
status_code=status.HTTP_201_CREATED,
summary="Create a new student",
)
def create_student(payload: StudentCreate, db: DbSession):
if crud.get_student_by_email(db, payload.email) is not None:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"A student with email {payload.email} already exists",
)
try:
return crud.create_student(db, payload)
except IntegrityError:
db.rollback()
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"A student with email {payload.email} already exists",
)
Reading it
@router.post("") An empty path on a router with prefix /students means the route is exactly POST /students.
The pre-check Look the email up first and return a friendly 409. This handles the normal case.
except IntegrityError The pre-check has a tiny gap: two requests with the same new email arriving in the same millisecond can both pass the check. Then both try to INSERT, and the database's unique=True rejects the second one by raising IntegrityError. We catch that, roll back the failed session (required; a session that hit an error cannot be reused until rolled back), and return the same 409. The database rule is the real guarantee; the pre-check is just for a nicer message.
Test
POST /students with Isha's body → 201, "id": 11, "grade": "B". Again → 409. In psql: SELECT * FROM students WHERE id = 11; shows the row.
Part 5: Updating and Deleting
The CRUD functions
src/studydesk/crud.py
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from .models import Student
from .schemas import StudentCreate
def get_student(db: Session, student_id: int) -> Student | None:
"""Return one student by primary key, or None."""
return db.get(Student, student_id)
def get_student_by_email(db: Session, email: str) -> Student | None:
"""Return the student with this email (case-insensitive), or None."""
stmt = select(Student).where(func.lower(Student.email) == email.lower())
return db.scalars(stmt).first()
def create_student(db: Session, payload: StudentCreate) -> Student:
"""Insert a new student and return the saved row (with its id)."""
student = Student(**payload.model_dump())
db.add(student)
db.commit()
db.refresh(student)
return student
def update_student(db: Session, student: Student, changes: dict) -> Student:
"""Apply a dict of field changes to an existing row and save."""
for field, value in changes.items():
setattr(student, field, value)
db.commit()
db.refresh(student)
return student
def delete_student(db: Session, student: Student) -> None:
"""Remove a row."""
db.delete(student)
db.commit()
Reading it
update_student(db, student, changes) Takes the row that the router already fetched (and already confirmed exists), plus the exclude_unset dict from the schema.
setattr(student, field, value) New built-in. setattr(obj, "marks", 91) is the same as obj.marks = 91, but with the attribute name coming from a variable. Since changes is a dict like {"marks": 91, "city": "Surat"}, the loop sets each one. SQLAlchemy notices the attributes changed and sends an UPDATE on commit.
db.delete(student) → db.commit() Mark for deletion, then check out. The DELETE runs on commit.
The routes
src/studydesk/routers/students.py
from typing import Annotated
from fastapi import APIRouter, HTTPException, Path, status
from sqlalchemy.exc import IntegrityError
from .. import crud
from ..database import DbSession
from ..schemas import StudentCreate, StudentOut, StudentUpdate
router = APIRouter(prefix="/students", tags=["Students"])
StudentId = Annotated[
int,
Path(
title="Student ID",
description="The unique ID of the student. Must be a positive integer.",
gt=0,
examples=[1],
),
]
def get_student_or_404(db: DbSession, student_id: StudentId):
"""Shared dependency: load the student or raise 404."""
student = crud.get_student(db, student_id)
if student is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Student with id {student_id} not found",
)
return student
@router.get("/{student_id}", response_model=StudentOut, summary="Get one student by ID")
def get_student(student=Depends(get_student_or_404)):
return student
@router.post(
"",
response_model=StudentOut,
status_code=status.HTTP_201_CREATED,
summary="Create a new student",
)
def create_student(payload: StudentCreate, db: DbSession):
if crud.get_student_by_email(db, payload.email) is not None:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"A student with email {payload.email} already exists",
)
try:
return crud.create_student(db, payload)
except IntegrityError:
db.rollback()
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"A student with email {payload.email} already exists",
)
@router.put("/{student_id}", response_model=StudentOut, summary="Update a student (partial)")
def update_student(
payload: StudentUpdate,
db: DbSession,
student=Depends(get_student_or_404),
):
changes = payload.model_dump(exclude_unset=True)
if not changes:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="No fields provided to update",
)
if "email" in changes:
other = crud.get_student_by_email(db, changes["email"])
if other is not None and other.id != student.id:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"A student with email {changes['email']} already exists",
)
try:
return crud.update_student(db, student, changes)
except IntegrityError:
db.rollback()
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"A student with email {changes.get('email')} already exists",
)
@router.delete(
"/{student_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="Delete a student",
)
def delete_student(db: DbSession, student=Depends(get_student_or_404)):
crud.delete_student(db, student)
One import is missing at the top: Depends. The correct first import line is:
from fastapi import APIRouter, Depends, HTTPException, Path, status
Reading the new parts
StudentId alias The long Annotated[int, Path(...)] block was about to be repeated in three routes. Same trick as DbSession: define it once, use it as a type.
get_student_or_404 as a dependency Three routes (get, update, delete) all begin with "load the student, 404 if missing". That is itself a dependency. It takes db and student_id (both resolved by FastAPI: one from get_db, one from the path), and either returns the row or raises. Routes then just write student=Depends(get_student_or_404) and receive a guaranteed-existing Student. The 404 logic lives in exactly one place.
Note what FastAPI did: the dependency declares student_id: StudentId, so FastAPI knows the route has a {student_id} path parameter, validates it (gt=0, 422 on abc), and documents it, even though the route function itself never mentions student_id. Dependencies can carry parameters.
Update: email uniqueness, excluding self other.id != student.id is the database version of exclude_id from Article 5.
Delete returns nothing crud.delete_student returns None, the route returns None, status is 204, body is empty.
Test
PUT /students/4 with {"marks": 91} → 200, grade A, other fields untouched. PUT /students/4 with {} → 400. PUT /students/4 with {"email": "priya@example.com"} → 409. DELETE /students/11 → 204. Again → 404.
Part 6: Fetch All — Filters, Sorting, and Pagination in SQL
The last and biggest route. In Article 3 we filtered, sorted, and sliced a Python list. Now the database does all of it, which is the whole point: PostgreSQL filters 30,000 rows faster than Python can load them.
Grade is not a column
One wrinkle. The client filters by grade=A, but the table has no grade column; grade is computed from marks. So in SQL, "grade A" must become "marks ≥ 85". We translate each grade into a marks range.
The CRUD function
src/studydesk/crud.py
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from .models import Student
from .schemas import StudentCreate
GRADE_RANGES = {
"A": (85, 100),
"B": (70, 84),
"C": (50, 69),
"D": (0, 49),
}
def get_student(db: Session, student_id: int) -> Student | None:
"""Return one student by primary key, or None."""
return db.get(Student, student_id)
def get_student_by_email(db: Session, email: str) -> Student | None:
"""Return the student with this email (case-insensitive), or None."""
stmt = select(Student).where(func.lower(Student.email) == email.lower())
return db.scalars(stmt).first()
def list_students(
db: Session,
*,
city: str | None = None,
grade: str | None = None,
min_marks: int | None = None,
sort_by: str = "id",
order: str = "asc",
offset: int = 0,
limit: int = 10,
) -> tuple[list[Student], int]:
"""Return (page of students, total matching count)."""
stmt = select(Student)
# --- filtering ---
if city:
stmt = stmt.where(func.lower(Student.city) == city.lower())
if grade:
low, high = GRADE_RANGES[grade.upper()]
stmt = stmt.where(Student.marks.between(low, high))
if min_marks is not None:
stmt = stmt.where(Student.marks >= min_marks)
# --- total count (before sorting and slicing) ---
total = db.scalar(select(func.count()).select_from(stmt.subquery()))
# --- sorting ---
column = getattr(Student, sort_by)
stmt = stmt.order_by(column.desc() if order == "desc" else column.asc())
# --- pagination ---
stmt = stmt.offset(offset).limit(limit)
return list(db.scalars(stmt).all()), total
def create_student(db: Session, payload: StudentCreate) -> Student:
"""Insert a new student and return the saved row (with its id)."""
student = Student(**payload.model_dump())
db.add(student)
db.commit()
db.refresh(student)
return student
def update_student(db: Session, student: Student, changes: dict) -> Student:
"""Apply a dict of field changes to an existing row and save."""
for field, value in changes.items():
setattr(student, field, value)
db.commit()
db.refresh(student)
return student
def delete_student(db: Session, student: Student) -> None:
"""Remove a row."""
db.delete(student)
db.commit()
Reading list_students
*, in the parameter list Everything after the lone * must be passed by keyword: list_students(db, city="Surat"), never list_students(db, "Surat"). With seven optional parameters, positional calls would be unreadable and error-prone, so Python lets us forbid them.
Building the statement step by step stmt = select(Student) starts as "all students". Each if adds a .where(...). Nothing is sent to the database yet; we are composing SQL. Same filter → sort → paginate order as Article 3.
Student.marks.between(low, high) SQL BETWEEN, inclusive on both ends. GRADE_RANGES["A"] is (85, 100), so grade A becomes marks BETWEEN 85 AND 100.
low, high = GRADE_RANGES[...] Tuple unpacking from the Python series.
The count
total = db.scalar(select(func.count()).select_from(stmt.subquery()))
We need "how many match the filters" before cutting a page. stmt.subquery() wraps the filtered statement so we can count its rows: SELECT count(*) FROM (SELECT ... WHERE ...). This runs as its own query.
getattr(Student, sort_by) getattr(obj, "marks") is the same as obj.marks, with the name coming from a variable. It is the read-twin of setattr from Part 5. Here it turns the string "marks" into the column object Student.marks. Because the router restricts sort_by with Literal, it can only ever be one of our four real columns.
column.desc() / column.asc() → SQL ORDER BY marks DESC.
.offset(offset).limit(limit) → SQL OFFSET 20 LIMIT 10. The database returns only the page.
return list(...), total A tuple: the page and the total. The router unpacks it.
The route
src/studydesk/routers/students.py
from typing import Annotated, Literal
from fastapi import APIRouter, Depends, HTTPException, Path, Query, status
from sqlalchemy.exc import IntegrityError
from .. import crud
from ..database import DbSession
from ..schemas import StudentCreate, StudentListOut, StudentOut, StudentUpdate
router = APIRouter(prefix="/students", tags=["Students"])
StudentId = Annotated[
int,
Path(
title="Student ID",
description="The unique ID of the student. Must be a positive integer.",
gt=0,
examples=[1],
),
]
def get_student_or_404(db: DbSession, student_id: StudentId):
"""Shared dependency: load the student or raise 404."""
student = crud.get_student(db, student_id)
if student is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Student with id {student_id} not found",
)
return student
@router.get("", response_model=StudentListOut, summary="Get all students")
def list_students(
db: DbSession,
city: Annotated[
str | None,
Query(description="Filter by city (case-insensitive).", examples=["Ahmedabad"]),
] = None,
grade: Annotated[
str | None,
Query(description="Filter by grade: A, B, C, or D.", pattern="^[ABCDabcd]$", examples=["A"]),
] = None,
min_marks: Annotated[
int | None,
Query(description="Only students with marks >= this value.", ge=0, le=100),
] = None,
sort_by: Annotated[
Literal["id", "name", "age", "marks"],
Query(description="Field to sort by."),
] = "id",
order: Annotated[
Literal["asc", "desc"],
Query(description="Sort direction: asc or desc."),
] = "asc",
offset: Annotated[int, Query(description="Number of students to skip.", ge=0)] = 0,
limit: Annotated[
int,
Query(description="Maximum number of students to return (1-100).", ge=1, le=100),
] = 10,
):
students, total = crud.list_students(
db,
city=city,
grade=grade,
min_marks=min_marks,
sort_by=sort_by,
order=order,
offset=offset,
limit=limit,
)
total_pages = (total + limit - 1) // limit if total > 0 else 0
return StudentListOut(
total=total,
total_pages=total_pages,
offset=offset,
limit=limit,
count=len(students),
students=students,
)
@router.get("/{student_id}", response_model=StudentOut, summary="Get one student by ID")
def get_student(student=Depends(get_student_or_404)):
return student
@router.post(
"",
response_model=StudentOut,
status_code=status.HTTP_201_CREATED,
summary="Create a new student",
)
def create_student(payload: StudentCreate, db: DbSession):
if crud.get_student_by_email(db, payload.email) is not None:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"A student with email {payload.email} already exists",
)
try:
return crud.create_student(db, payload)
except IntegrityError:
db.rollback()
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"A student with email {payload.email} already exists",
)
@router.put("/{student_id}", response_model=StudentOut, summary="Update a student (partial)")
def update_student(
payload: StudentUpdate,
db: DbSession,
student=Depends(get_student_or_404),
):
changes = payload.model_dump(exclude_unset=True)
if not changes:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="No fields provided to update",
)
if "email" in changes:
other = crud.get_student_by_email(db, changes["email"])
if other is not None and other.id != student.id:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"A student with email {changes['email']} already exists",
)
try:
return crud.update_student(db, student, changes)
except IntegrityError:
db.rollback()
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail=f"A student with email {changes.get('email')} already exists",
)
@router.delete(
"/{student_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="Delete a student",
)
def delete_student(db: DbSession, student=Depends(get_student_or_404)):
crud.delete_student(db, student)
Reading the list route
Route order matters GET "" (the list) is declared before GET "/{student_id}". FastAPI matches routes top to bottom. This particular pair cannot collide (/students vs /students/3), but as a habit, put fixed paths before parameterised ones so that a fixed path like /students/search is never swallowed by /{student_id}.
response_model=StudentListOut The paginated envelope from Article 3, now a typed schema. students: list[StudentOut] inside it means every row in the list goes through StudentOut, gets its grade, and is filtered to the public fields. /docs shows the full nested shape.
return StudentListOut(...) We build the schema explicitly here, because the function has to assemble several values (the list plus the counts). Passing SQLAlchemy rows into students= works because of from_attributes.
pattern="^[ABCDabcd]$" Small improvement: accept lowercase grade letters too; crud upper-cases before the lookup.
Test
/students?limit=3 → total: 10, total_pages: 4, count: 3, IDs 1–3. /students?grade=B&sort_by=marks&order=desc&limit=2 → Yash 84, Kavya 81; total: 4. /students?city=surat&min_marks=80 → Yash only. /students?sort_by=email → 422.
Switch echo=True in database.py for a moment and watch the terminal: the filters, ORDER BY, OFFSET, and LIMIT are all in the SQL that PostgreSQL receives. Python never saw the rows that were filtered out. Switch it back to False.
Part 7: Where the Logic Lives Now
| Layer | File | Knows about |
|---|---|---|
| HTTP | routers/students.py |
paths, query params, status codes, HTTPException |
| Data access | crud.py |
sessions, select, commit — no HTTP |
| Shapes | schemas.py |
validation in, response_model out |
| Tables | models.py |
columns, constraints |
| Plumbing | database.py, config.py |
engine, sessions, .env |
A route never touches SQL. A CRUD function never raises an HTTPException. This split is what will let Article 13 test crud.py against a throwaway database with no HTTP involved at all.
Where StudyDesk is after this article
studydesk/
└── src/
└── studydesk/
├── __init__.py ← app, lifespan, CORS, include_router, /, /db-check
├── config.py
├── crud.py ← get, get_by_email, list, create, update, delete
├── database.py ← engine, SessionLocal, Base, get_db, DbSession
├── models.py ← SQLAlchemy Student
├── schemas.py ← StudentCreate, StudentUpdate, StudentOut, StudentListOut
├── seed.py
├── routers/
│ ├── __init__.py
│ └── students.py ← all five student routes
└── data/students.json ← only used by seed.py
utils.py is deleted. Every route is backed by PostgreSQL, and every response is typed.
| Method | Path | response_model | Success |
|---|---|---|---|
| GET | /students |
StudentListOut |
200 |
| POST | /students |
StudentOut |
201 |
| GET | /students/{id} |
StudentOut |
200 |
| PUT | /students/{id} |
StudentOut |
200 |
| DELETE | /students/{id} |
— | 204 |
Concept Quiz
Answers are below. No coding needed.
Q1. What does APIRouter(prefix="/students") change about the paths declared on that router?
Q2. Name the four things response_model does for a route.
Q3. Why must routes not raise HTTPException inside crud.py functions?
Q4. What does db.refresh(student) do after a commit, and why is it needed?
Q5. The create route checks the email first and catches IntegrityError. Why both?
Q6. Why must db.rollback() be called after catching IntegrityError?
Q7. What do setattr(student, "marks", 91) and getattr(Student, "marks") each do?
Q8. How does get_student_or_404 as a dependency cause FastAPI to know about the {student_id} path parameter?
Q9. Why is grade filtering done with marks BETWEEN low AND high instead of a grade column?
Q10. What does the lone * in def list_students(db, *, city=None, ...) enforce?
Answers
- Every path on the router gets
/studentsprepended:""becomes/students,"/{student_id}"becomes/students/{student_id}. - Converts the returned object (a SQLAlchemy row) into the schema via
from_attributes; adds computed fields likegrade; filters out any fields not in the schema; documents the exact response shape in/docs. crud.pyis the data-access layer and should know nothing about HTTP. The router decides what aNoneresult means (404). This keepscrudreusable and independently testable.- It reloads the row from the database so database-generated values, especially the new
id, are present on the object. Without it the id may be missing. - The pre-check gives a friendly 409 in the normal case. The
IntegrityErrorcatch covers the race where two requests pass the check simultaneously; the database'suniqueconstraint is the real guarantee. - A session that raised a database error is in a failed state and cannot run further statements until it is rolled back.
setattrassigns an attribute whose name comes from a string (student.marks = 91).getattrreads an attribute whose name comes from a string (Student.marks, the column object).- FastAPI inspects the dependency's parameters.
student_id: StudentIdcarriesPath(...)metadata, so FastAPI treats it as a path parameter, validates it, and documents it, even though the route function does not list it. - There is no
gradecolumn; grade is computed frommarksin the output schema. To filter in SQL, each grade letter is translated into its marks range. - Everything after it must be passed by keyword, which prevents confusing positional calls with many optional parameters.
Epilogue
Aman runs Karan's frontend against the new routes. Nothing breaks. The JSON is byte-for-byte the same. The difference is in the terminal: every request is now a tidy SELECT, INSERT, UPDATE, or DELETE, and Ritu and Pavan can click at the same second without losing data.
Neha says, "It runs on your laptop with your PostgreSQL. Tomorrow it has to run on a server that has nothing installed, and on Karan's laptop, identically. Let's put the whole thing in a box."
In the next article: Docker and Docker Compose — a Dockerfile for the API, a Compose file that starts the API and PostgreSQL together, and one command to bring StudyDesk up anywhere.
Comments
Join the discussionNo comments yet. Be the first to comment.