Python 19 - Students and Courses: SQLAlchemy Relationships and Transactions

Connect tables to each other, load related data without flooding the database, and make several writes succeed or fail together.

Where StudyDesk is now

src/studydesk/
├── models.py            ← Student, User, Job (no links between them)
├── crud.py, schemas.py, routers/...
└── ...
alembic/versions/        ← baseline, add phone
tests/                   ← 49 tests

Every table stands alone. By the end of this article, StudyDesk has courses and enrollments tables, a student can be enrolled in many courses and a course can hold many students, enrolling is an all-or-nothing transaction that respects seat limits, and deleting a student cleans up their enrollments automatically.

Part 1: Two Clerks, One Seat

The institute runs courses: "Python Basics", 20 seats; "Data Science", 15 seats. Students enrol. The API needs to:

create courses
enrol a student in a course, only if a seat is free
list a course's students, and a student's courses
remove an enrollment

Aman writes the enrol logic: read the course, check seats_taken < seats, insert an enrollment, add one to seats_taken, commit.

Neha's test: course with one seat left, two clerks click "Enrol" for two different students at the same instant. Both reads see one free seat. Both insert. Both increment. The course now has 21 students in 20 seats.

"Two problems today," she says. "How tables relate to each other, and how to make several writes behave as one — so that the second clerk is told no seat instead of overbooking."

Part 2: Relationships in a Database

Foreign keys

A foreign key is a column that holds the primary key of a row in another table. It is how the database records "this row belongs to that one".

enrollments.student_id  →  students.id
enrollments.course_id   →  courses.id

The database enforces it: you cannot insert an enrollment pointing at a student that does not exist, and (depending on settings) you cannot delete a student who still has enrollments, or the deletion cascades to them.

The three shapes

Shape	Example	How
One-to-many	one course has many enrollments	a foreign key on the "many" side (`enrollments.course_id`)
Many-to-one	each enrollment belongs to one course	the same foreign key, seen from the other end
Many-to-many	a student has many courses, a course has many students	a link table in the middle (`enrollments`), with one foreign key to each side

Students and courses are many-to-many. The link table is enrollments, and because we want to store when someone enrolled, it is a real table with its own columns, not just two foreign keys. SQLAlchemy calls this an association object.

students ──< enrollments >── courses
   1           many  many        1

Part 3: The Models

src/studydesk/models.py

from datetime import datetime

from sqlalchemy import JSON, Boolean, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from .database import Base


class Student(Base):
    __tablename__ = "students"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50), nullable=False)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
    phone: Mapped[str | None] = mapped_column(String(20), nullable=True)
    age: Mapped[int] = mapped_column(Integer, nullable=False)
    city: Mapped[str] = mapped_column(String(50), nullable=False)
    marks: Mapped[int] = mapped_column(Integer, nullable=False)

    enrollments: Mapped[list["Enrollment"]] = relationship(
        back_populates="student", cascade="all, delete-orphan"
    )

    def __repr__(self) -> str:
        return f"<Student id={self.id} name={self.name!r} marks={self.marks}>"


class Course(Base):
    __tablename__ = "courses"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    code: Mapped[str] = mapped_column(String(20), unique=True, nullable=False, index=True)
    title: Mapped[str] = mapped_column(String(100), nullable=False)
    seats: Mapped[int] = mapped_column(Integer, nullable=False)
    seats_taken: Mapped[int] = mapped_column(Integer, nullable=False, default=0)

    enrollments: Mapped[list["Enrollment"]] = relationship(
        back_populates="course", cascade="all, delete-orphan"
    )

    def __repr__(self) -> str:
        return f"<Course id={self.id} code={self.code} seats={self.seats_taken}/{self.seats}>"


class Enrollment(Base):
    __tablename__ = "enrollments"
    __table_args__ = (UniqueConstraint("student_id", "course_id", name="uq_enrollment_student_course"),)

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    student_id: Mapped[int] = mapped_column(
        ForeignKey("students.id", ondelete="CASCADE"), nullable=False, index=True
    )
    course_id: Mapped[int] = mapped_column(
        ForeignKey("courses.id", ondelete="CASCADE"), nullable=False, index=True
    )
    enrolled_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )

    student: Mapped["Student"] = relationship(back_populates="enrollments")
    course: Mapped["Course"] = relationship(back_populates="enrollments")

    def __repr__(self) -> str:
        return f"<Enrollment student={self.student_id} course={self.course_id}>"


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    role: Mapped[str] = mapped_column(String(20), nullable=False, default="staff")
    is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )

    def __repr__(self) -> str:
        return f"<User id={self.id} email={self.email!r} role={self.role}>"


class Job(Base):
    __tablename__ = "jobs"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    kind: Mapped[str] = mapped_column(String(50), nullable=False, index=True)
    payload: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict)
    status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending", index=True)
    result: Mapped[dict | None] = mapped_column(JSON, nullable=True)
    error: Mapped[str | None] = mapped_column(Text, nullable=True)
    attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    max_attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=3)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), nullable=False, server_default=func.now()
    )
    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

    def __repr__(self) -> str:
        return f"<Job id={self.id} kind={self.kind} status={self.status}>"

Reading the new parts

ForeignKey("students.id", ondelete="CASCADE") student_id must match an existing students.id. ondelete="CASCADE" tells PostgreSQL: when a student row is deleted, delete its enrollment rows too. This is enforced at the database level, regardless of which code does the delete.

relationship(...) This is the SQLAlchemy side. A relationship is not a column; it is a Python attribute that loads related rows for you. course.enrollments gives a list of Enrollment objects; enrollment.student gives the Student. SQLAlchemy figures out the join from the foreign keys.

back_populates Links the two ends so they stay in sync in memory. Student.enrollments says back_populates="student", and Enrollment.student says back_populates="enrollments". Append an enrollment to student.enrollments and enrollment.student is set automatically.

cascade="all, delete-orphan" The ORM side of cascading. When you db.delete(student), SQLAlchemy also deletes the enrollments it has loaded. Together with ondelete="CASCADE" on the column, both layers agree.

Mapped[list["Enrollment"]] with the name in quotes Enrollment is defined later in the file. The string lets Python reference a class that does not exist yet; SQLAlchemy resolves it when all models are loaded.

UniqueConstraint("student_id", "course_id") A student may be enrolled in a course only once. A multi-column unique rule goes in __table_args__.

seats_taken We store the count instead of counting enrollments on every read. It is faster, and it is the thing the transaction will protect.

The migration

uv run alembic revision --autogenerate -m "add courses and enrollments"

INFO  [alembic.autogenerate.compare] Detected added table 'courses'
INFO  [alembic.autogenerate.compare] Detected added table 'enrollments'

alembic/versions/c3d4e5f6a7b8_add_courses_and_enrollments.py

"""add courses and enrollments

Revision ID: c3d4e5f6a7b8
Revises: b7c8d9e0f1a2
Create Date: 2026-10-31 23:30:00.000000

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


revision: str = "c3d4e5f6a7b8"
down_revision: Union[str, None] = "b7c8d9e0f1a2"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    op.create_table(
        "courses",
        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
        sa.Column("code", sa.String(length=20), nullable=False),
        sa.Column("title", sa.String(length=100), nullable=False),
        sa.Column("seats", sa.Integer(), nullable=False),
        sa.Column("seats_taken", sa.Integer(), nullable=False),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_index(op.f("ix_courses_code"), "courses", ["code"], unique=True)

    op.create_table(
        "enrollments",
        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
        sa.Column("student_id", sa.Integer(), nullable=False),
        sa.Column("course_id", sa.Integer(), nullable=False),
        sa.Column("enrolled_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
        sa.ForeignKeyConstraint(["course_id"], ["courses.id"], ondelete="CASCADE"),
        sa.ForeignKeyConstraint(["student_id"], ["students.id"], ondelete="CASCADE"),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("student_id", "course_id", name="uq_enrollment_student_course"),
    )
    op.create_index(op.f("ix_enrollments_course_id"), "enrollments", ["course_id"], unique=False)
    op.create_index(op.f("ix_enrollments_student_id"), "enrollments", ["student_id"], unique=False)


def downgrade() -> None:
    op.drop_index(op.f("ix_enrollments_student_id"), table_name="enrollments")
    op.drop_index(op.f("ix_enrollments_course_id"), table_name="enrollments")
    op.drop_table("enrollments")
    op.drop_index(op.f("ix_courses_code"), table_name="courses")
    op.drop_table("courses")

Autogenerate got the foreign keys, the cascade, and the unique constraint. Apply:

uv run alembic upgrade head

Part 4: Schemas

src/studydesk/schemas.py — add this section after the jobs section (the rest of the file is unchanged from Article 12)

# ---------- courses and enrollments ----------


class CourseCreate(BaseModel):
    """Input schema for POST /courses."""

    code: str = Field(min_length=2, max_length=20, pattern=r"^[A-Z0-9-]+$", examples=["PY-101"])
    title: str = Field(min_length=3, max_length=100, examples=["Python Basics"])
    seats: int = Field(ge=1, le=500, examples=[20])


class CourseOut(BaseModel):
    """Output schema for a course."""

    model_config = ConfigDict(from_attributes=True)

    id: int
    code: str
    title: str
    seats: int
    seats_taken: int

    @computed_field
    @property
    def seats_available(self) -> int:
        return self.seats - self.seats_taken


class EnrollmentOut(BaseModel):
    """Output schema for one enrollment, with the student embedded."""

    model_config = ConfigDict(from_attributes=True)

    id: int
    enrolled_at: datetime
    student: StudentOut


class CourseWithStudentsOut(CourseOut):
    """A course plus everyone enrolled in it."""

    enrollments: list[EnrollmentOut]


class StudentCourseOut(BaseModel):
    """One course from a student's point of view."""

    model_config = ConfigDict(from_attributes=True)

    id: int
    enrolled_at: datetime
    course: CourseOut

Nested schemas EnrollmentOut contains a StudentOut. Because of from_attributes=True, Pydantic follows enrollment.student (the relationship) and builds the nested object. CourseWithStudentsOut contains a list of those. One response_model can describe a whole tree.

seats_available — computed, never stored, same as grade.

Before the CRUD, one trap to understand.

Lazy loading

By default a relationship is lazy: course.enrollments is not fetched when the course is loaded. It is fetched the first time you touch it, with a separate query. Then enrollment.student is another query, per enrollment.

So GET /courses/1 with 20 students, written naively, runs:

1 query for the course
1 query for its enrollments
20 queries, one per student

Twenty-two round trips to show one course. With 200 students, 202. This is the N+1 problem, and it is the most common performance bug in ORM code.

Eager loading with `selectinload`

Tell SQLAlchemy up front what you will need:

select(Course)
    .options(selectinload(Course.enrollments).selectinload(Enrollment.student))
    .where(Course.id == course_id)

selectinload fetches the related rows in one extra query per relationship level using WHERE id IN (...). Three queries total, however many students the course has. Switch echo=True in database.py once and compare; you will never forget it.

Part 6: Transactions

What a transaction is

A transaction is a group of database operations that either all happen or none happen. Between BEGIN and COMMIT, nothing is visible to anyone else; if anything fails, ROLLBACK undoes every step.

Enrolling is three operations:

check the course has a free seat
insert the enrollment row
add one to seats_taken

If step 3 fails after step 2 succeeded, the course shows one fewer seat taken than it really has. If two clerks run step 1 at the same time, they both pass. A transaction fixes the first problem. A row lock inside the transaction fixes the second.

Sessions already use transactions

Since autocommit=False, every session is in a transaction from its first query until db.commit() or db.rollback(). Every CRUD function so far has been one small transaction. Enrolling just needs to put three steps inside the same one, and lock the course row for its duration.

`with_for_update()`

course = db.scalars(select(Course).where(Course.id == course_id).with_for_update()).first()

SELECT ... FOR UPDATE locks the row until the transaction ends. A second transaction that tries to lock the same row waits. When the first commits, the second proceeds, re-reads the row, sees seats_taken is now 20, and fails the seat check. No overbooking.

(You met with_for_update(skip_locked=True) in the job queue, where waiting workers skip to the next row. Here we want them to wait for the same row, so no skip_locked.)

Part 7: CRUD for Courses and Enrollments

src/studydesk/crud.py

from datetime import datetime, timezone

from sqlalchemy import func, select
from sqlalchemy.orm import Session, selectinload

from .models import Course, Enrollment, Job, Student, User
from .schemas import CourseCreate, StudentCreate, UserCreate
from .security import hash_password, verify_password

GRADE_RANGES = {
    "A": (85, 100),
    "B": (70, 84),
    "C": (50, 69),
    "D": (0, 49),
}


class EnrollmentError(Exception):
    """Raised when an enrollment cannot be made. The message says why."""


# ---------- students ----------


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)

    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 = db.scalar(select(func.count()).select_from(stmt.subquery()))

    column = getattr(Student, sort_by)
    stmt = stmt.order_by(column.desc() if order == "desc" else column.asc())

    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. Enrollments are removed by cascade."""
    db.delete(student)
    db.commit()


def list_student_courses(db: Session, student_id: int) -> list[Enrollment]:
    """All enrollments of one student, with each course loaded."""
    stmt = (
        select(Enrollment)
        .options(selectinload(Enrollment.course))
        .where(Enrollment.student_id == student_id)
        .order_by(Enrollment.enrolled_at)
    )
    return list(db.scalars(stmt).all())


# ---------- courses ----------


def get_course(db: Session, course_id: int) -> Course | None:
    return db.get(Course, course_id)


def get_course_with_students(db: Session, course_id: int) -> Course | None:
    """One course with its enrollments and each enrolled student, in 3 queries."""
    stmt = (
        select(Course)
        .options(selectinload(Course.enrollments).selectinload(Enrollment.student))
        .where(Course.id == course_id)
    )
    return db.scalars(stmt).first()


def get_course_by_code(db: Session, code: str) -> Course | None:
    stmt = select(Course).where(Course.code == code.upper())
    return db.scalars(stmt).first()


def list_courses(db: Session) -> list[Course]:
    return list(db.scalars(select(Course).order_by(Course.code)).all())


def create_course(db: Session, payload: CourseCreate) -> Course:
    course = Course(code=payload.code.upper(), title=payload.title, seats=payload.seats)
    db.add(course)
    db.commit()
    db.refresh(course)
    return course


# ---------- enrollments (transactional) ----------


def enroll_student(db: Session, course_id: int, student_id: int) -> Enrollment:
    """
    Enrol a student in a course as ONE transaction:
      lock the course row → check a seat is free → insert enrollment → increment seats_taken → commit.
    Raises EnrollmentError (and rolls back) if any rule fails.
    """
    try:
        course = db.scalars(
            select(Course).where(Course.id == course_id).with_for_update()
        ).first()
        if course is None:
            raise EnrollmentError(f"Course {course_id} not found")

        student = db.get(Student, student_id)
        if student is None:
            raise EnrollmentError(f"Student {student_id} not found")

        already = db.scalars(
            select(Enrollment).where(
                Enrollment.course_id == course_id,
                Enrollment.student_id == student_id,
            )
        ).first()
        if already is not None:
            raise EnrollmentError(f"Student {student_id} is already enrolled in {course.code}")

        if course.seats_taken >= course.seats:
            raise EnrollmentError(f"Course {course.code} is full ({course.seats} seats)")

        enrollment = Enrollment(student_id=student_id, course_id=course_id)
        db.add(enrollment)
        course.seats_taken += 1

        db.commit()
    except Exception:
        db.rollback()
        raise

    db.refresh(enrollment)
    return enrollment


def unenroll_student(db: Session, course_id: int, student_id: int) -> bool:
    """Remove an enrollment and free the seat, as one transaction. Returns False if not enrolled."""
    try:
        course = db.scalars(
            select(Course).where(Course.id == course_id).with_for_update()
        ).first()
        if course is None:
            return False

        enrollment = db.scalars(
            select(Enrollment).where(
                Enrollment.course_id == course_id,
                Enrollment.student_id == student_id,
            )
        ).first()
        if enrollment is None:
            db.rollback()
            return False

        db.delete(enrollment)
        course.seats_taken -= 1
        db.commit()
        return True
    except Exception:
        db.rollback()
        raise


# ---------- users ----------


def get_user_by_email(db: Session, email: str) -> User | None:
    """Return the user with this email (case-insensitive), or None."""
    stmt = select(User).where(func.lower(User.email) == email.lower())
    return db.scalars(stmt).first()


def create_user(db: Session, payload: UserCreate, role: str = "staff") -> User:
    """Insert a new user with a hashed password."""
    user = User(
        email=payload.email.lower(),
        hashed_password=hash_password(payload.password),
        role=role,
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return user


def authenticate_user(db: Session, email: str, password: str) -> User | None:
    """Return the user if the email exists and the password matches, else None."""
    user = get_user_by_email(db, email)
    if user is None:
        return None
    if not verify_password(password, user.hashed_password):
        return None
    return user


# ---------- jobs ----------


def create_job(db: Session, kind: str, payload: dict | None = None) -> Job:
    """Enqueue a job: insert a pending row."""
    job = Job(kind=kind, payload=payload or {})
    db.add(job)
    db.commit()
    db.refresh(job)
    return job


def get_job(db: Session, job_id: int) -> Job | None:
    return db.get(Job, job_id)


def claim_next_job(db: Session) -> Job | None:
    """Atomically pick one pending job and mark it running. Returns None if the queue is empty."""
    stmt = (
        select(Job)
        .where(Job.status == "pending")
        .order_by(Job.created_at)
        .limit(1)
        .with_for_update(skip_locked=True)
    )
    job = db.scalars(stmt).first()
    if job is None:
        db.rollback()
        return None

    job.status = "running"
    job.attempts += 1
    job.started_at = datetime.now(timezone.utc)
    db.commit()
    db.refresh(job)
    return job


def finish_job(db: Session, job: Job, result: dict) -> None:
    job.status = "done"
    job.result = result
    job.error = None
    job.finished_at = datetime.now(timezone.utc)
    db.commit()


def fail_job(db: Session, job: Job, error: str) -> None:
    """Mark as failed, or back to pending if attempts remain."""
    job.error = error
    job.finished_at = datetime.now(timezone.utc)
    job.status = "pending" if job.attempts < job.max_attempts else "failed"
    db.commit()

Reading `enroll_student`

class EnrollmentError(Exception) A custom exception (inheritance from the Python series). crud.py still knows nothing about HTTP; it raises this, and the router turns it into the right status code. One exception type, several messages.

The order inside the transaction

Lock the course first, with with_for_update(). Everything after this line runs while no other transaction can touch this course row.
Check the student exists.
Check not already enrolled (the unique constraint would also catch this, but a clear message is better).
Check seats — reading seats_taken after the lock, so the value is current.
Insert and increment. Both are only staged in the session.
db.commit() — both writes land together and the lock is released.

except Exception: db.rollback(); raise Any failure — our own EnrollmentError, an IntegrityError, a lost connection — rolls back everything staged and releases the lock, then re-raises so the caller sees it. raise with nothing after it re-raises the current exception unchanged.

Replaying Neha's test Clerk A's transaction locks the course and reads 19 taken. Clerk B's transaction reaches with_for_update() and waits. A inserts, increments to 20, commits. B's lock is granted; B reads seats_taken — SQLAlchemy re-fetches the row under the lock — sees 20, hits seats_taken >= seats, raises EnrollmentError("... is full"), rolls back. Twenty students, twenty seats.

Reading `unenroll_student`

Same shape: lock the course, find the enrollment, delete it, decrement, commit. Returns False for "nothing to remove" so the router can send 404.

Part 8: Routers

Courses router

src/studydesk/routers/courses.py

from typing import Annotated

from fastapi import APIRouter, HTTPException, Path, status

from .. import crud
from ..database import DbSession
from ..dependencies import AdminUser, CurrentUser
from ..schemas import CourseCreate, CourseOut, CourseWithStudentsOut, EnrollmentOut

router = APIRouter(prefix="/courses", tags=["Courses"])

CourseId = Annotated[int, Path(gt=0, description="Course ID")]
StudentIdPath = Annotated[int, Path(gt=0, description="Student ID")]


@router.get("", response_model=list[CourseOut], summary="List all courses")
def list_courses(db: DbSession, user: CurrentUser):
    return crud.list_courses(db)


@router.post(
    "",
    response_model=CourseOut,
    status_code=status.HTTP_201_CREATED,
    summary="Create a course (admin only)",
)
def create_course(payload: CourseCreate, db: DbSession, admin: AdminUser):
    if crud.get_course_by_code(db, payload.code) is not None:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"A course with code {payload.code.upper()} already exists",
        )
    return crud.create_course(db, payload)


@router.get("/{course_id}", response_model=CourseWithStudentsOut, summary="Get a course with its students")
def get_course(course_id: CourseId, db: DbSession, user: CurrentUser):
    course = crud.get_course_with_students(db, course_id)
    if course is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Course {course_id} not found")
    return course


@router.post(
    "/{course_id}/students/{student_id}",
    response_model=EnrollmentOut,
    status_code=status.HTTP_201_CREATED,
    summary="Enrol a student in a course",
)
def enroll(course_id: CourseId, student_id: StudentIdPath, db: DbSession, user: CurrentUser):
    try:
        enrollment = crud.enroll_student(db, course_id, student_id)
    except crud.EnrollmentError as exc:
        message = str(exc)
        if "not found" in message:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=message)
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=message)

    db.refresh(enrollment, attribute_names=["student"])
    return enrollment


@router.delete(
    "/{course_id}/students/{student_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="Remove a student from a course",
)
def unenroll(course_id: CourseId, student_id: StudentIdPath, db: DbSession, user: CurrentUser):
    removed = crud.unenroll_student(db, course_id, student_id)
    if not removed:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Student {student_id} is not enrolled in course {course_id}",
        )

Reading it

Nested resource paths POST /courses/5/students/3 reads as "add student 3 to course 5". Two path parameters, both validated by Path(gt=0).

Mapping EnrollmentError to status codes "not found" → 404; anything else (full, already enrolled) → 409 Conflict. The router owns the HTTP decision; crud.py owns the rule.

db.refresh(enrollment, attribute_names=["student"]) After the commit, load the student relationship so EnrollmentOut can embed it. One targeted query instead of a lazy load during serialisation.

response_model=list[CourseOut] A list schema: each Course row becomes a CourseOut.

A student's courses

Add one route to the students router. Full file:

src/studydesk/routers/students.py

from typing import Annotated, Literal

from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Path, Query, status
from sqlalchemy.exc import IntegrityError

from .. import crud
from ..database import DbSession
from ..dependencies import AdminUser, CurrentUser
from ..schemas import StudentCourseOut, StudentCreate, StudentListOut, StudentOut, StudentUpdate
from ..tasks import send_welcome_email

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,
    user: CurrentUser,
    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(user: CurrentUser, student=Depends(get_student_or_404)):
    return student


@router.get(
    "/{student_id}/courses",
    response_model=list[StudentCourseOut],
    summary="List the courses a student is enrolled in",
)
def get_student_courses(db: DbSession, user: CurrentUser, student=Depends(get_student_or_404)):
    return crud.list_student_courses(db, student.id)


@router.post(
    "",
    response_model=StudentOut,
    status_code=status.HTTP_201_CREATED,
    summary="Create a new student",
)
def create_student(
    payload: StudentCreate,
    db: DbSession,
    user: CurrentUser,
    background_tasks: BackgroundTasks,
):
    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:
        student = 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",
        )

    background_tasks.add_task(send_welcome_email, student.email, student.name)
    return student


@router.put("/{student_id}", response_model=StudentOut, summary="Update a student (partial)")
def update_student(
    payload: StudentUpdate,
    db: DbSession,
    user: CurrentUser,
    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 (admin only)",
)
def delete_student(db: DbSession, admin: AdminUser, student=Depends(get_student_or_404)):
    crud.delete_student(db, student)

Route order /{student_id}/courses is declared right after /{student_id}. They do not collide (different segment counts), and keeping them together reads well.

Wire the router

src/studydesk/__init__.py

import logging
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, HOST, PORT, RELOAD
from .database import DbSession
from .models import Student
from .routers import auth, courses, jobs, students

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Schema is managed by Alembic (run `alembic upgrade head` before starting).
    yield


app = FastAPI(
    title="StudyDesk API",
    description="Student records API for a coaching institute.",
    version="0.6.0",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(auth.router)
app.include_router(students.router)
app.include_router(courses.router)
app.include_router(jobs.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 server. Host, port and reload come from the environment."""
    uvicorn.run("studydesk:app", host=HOST, port=PORT, reload=RELOAD)

Part 9: Walking Through It

Start the API, log in as admin, then:

Create a course. POST /courses with {"code": "py-101", "title": "Python Basics", "seats": 2} → 201:

{"id": 1, "code": "PY-101", "title": "Python Basics", "seats": 2, "seats_taken": 0, "seats_available": 2}

Enrol two students. POST /courses/1/students/1 → 201, with Priya embedded. POST /courses/1/students/2 → 201. GET /courses/1 → seats_taken: 2, seats_available: 0, two enrollments with full student objects.

Try a third. POST /courses/1/students/3 → 409, "Course PY-101 is full (2 seats)".

Try a duplicate. POST /courses/1/students/1 → 409, "Student 1 is already enrolled in PY-101".

From the student's side. GET /students/1/courses → a list with one entry containing the course.

Unenrol. DELETE /courses/1/students/2 → 204. GET /courses/1 → seats_available: 1. Now student 3 can enrol.

Cascade. As admin, DELETE /students/1 → 204. GET /courses/1 → Priya's enrollment is gone. (The seat count was not decremented here, because delete_student did not go through unenroll_student; see the quiz for the fix.)

Watch the lock. In psql, run:

BEGIN;
SELECT * FROM courses WHERE id = 1 FOR UPDATE;

Leave it open. Now POST /courses/1/students/3 from /docs: the request hangs. Back in psql: COMMIT;. The request completes. That pause is the lock doing its job.

Part 10: Tests

tests/test_courses.py

import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session

from studydesk import crud
from studydesk.schemas import CourseCreate, StudentCreate


@pytest.fixture(autouse=True)
def no_real_email(monkeypatch):
    monkeypatch.setattr("studydesk.routers.students.send_welcome_email", lambda *a: None)


@pytest.fixture
def course(client: TestClient, admin_headers) -> dict:
    response = client.post(
        "/courses", json={"code": "PY-101", "title": "Python Basics", "seats": 2}, headers=admin_headers
    )
    assert response.status_code == 201
    return response.json()


@pytest.fixture
def three_students(client: TestClient, staff_headers, sample_student) -> list[int]:
    ids = []
    for i in range(3):
        payload = dict(sample_student, email=f"s{i}@example.com")
        ids.append(client.post("/students", json=payload, headers=staff_headers).json()["id"])
    return ids


# ---------- courses ----------


def test_create_course_requires_admin(client: TestClient, staff_headers):
    response = client.post(
        "/courses", json={"code": "PY-101", "title": "Python Basics", "seats": 2}, headers=staff_headers
    )
    assert response.status_code == 403


def test_create_course_uppercases_code(course):
    assert course["code"] == "PY-101"
    assert course["seats_available"] == 2


def test_duplicate_course_code_is_409(client: TestClient, admin_headers, course):
    response = client.post(
        "/courses", json={"code": "py-101", "title": "Other", "seats": 5}, headers=admin_headers
    )
    assert response.status_code == 409


def test_list_courses(client: TestClient, staff_headers, course):
    response = client.get("/courses", headers=staff_headers)
    assert response.status_code == 200
    assert [c["code"] for c in response.json()] == ["PY-101"]


# ---------- enrollments ----------


def test_enroll_and_course_shows_student(client: TestClient, staff_headers, course, three_students):
    sid = three_students[0]
    response = client.post(f"/courses/{course['id']}/students/{sid}", headers=staff_headers)
    assert response.status_code == 201
    assert response.json()["student"]["id"] == sid

    detail = client.get(f"/courses/{course['id']}", headers=staff_headers).json()
    assert detail["seats_taken"] == 1
    assert detail["seats_available"] == 1
    assert [e["student"]["id"] for e in detail["enrollments"]] == [sid]


def test_enroll_full_course_is_409(client: TestClient, staff_headers, course, three_students):
    cid = course["id"]
    assert client.post(f"/courses/{cid}/students/{three_students[0]}", headers=staff_headers).status_code == 201
    assert client.post(f"/courses/{cid}/students/{three_students[1]}", headers=staff_headers).status_code == 201
    response = client.post(f"/courses/{cid}/students/{three_students[2]}", headers=staff_headers)
    assert response.status_code == 409
    assert "full" in response.json()["detail"]


def test_enroll_twice_is_409(client: TestClient, staff_headers, course, three_students):
    cid, sid = course["id"], three_students[0]
    assert client.post(f"/courses/{cid}/students/{sid}", headers=staff_headers).status_code == 201
    response = client.post(f"/courses/{cid}/students/{sid}", headers=staff_headers)
    assert response.status_code == 409
    assert "already enrolled" in response.json()["detail"]


def test_enroll_unknown_student_is_404(client: TestClient, staff_headers, course):
    response = client.post(f"/courses/{course['id']}/students/999", headers=staff_headers)
    assert response.status_code == 404


def test_enroll_unknown_course_is_404(client: TestClient, staff_headers, three_students):
    response = client.post(f"/courses/999/students/{three_students[0]}", headers=staff_headers)
    assert response.status_code == 404


def test_student_courses(client: TestClient, staff_headers, course, three_students):
    sid = three_students[0]
    client.post(f"/courses/{course['id']}/students/{sid}", headers=staff_headers)
    response = client.get(f"/students/{sid}/courses", headers=staff_headers)
    assert response.status_code == 200
    assert [e["course"]["code"] for e in response.json()] == ["PY-101"]


def test_unenroll_frees_seat(client: TestClient, staff_headers, course, three_students):
    cid, sid = course["id"], three_students[0]
    client.post(f"/courses/{cid}/students/{sid}", headers=staff_headers)
    assert client.delete(f"/courses/{cid}/students/{sid}", headers=staff_headers).status_code == 204
    assert client.get(f"/courses/{cid}", headers=staff_headers).json()["seats_taken"] == 0


def test_unenroll_not_enrolled_is_404(client: TestClient, staff_headers, course, three_students):
    response = client.delete(f"/courses/{course['id']}/students/{three_students[0]}", headers=staff_headers)
    assert response.status_code == 404


def test_deleting_student_removes_enrollment(client: TestClient, staff_headers, admin_headers, course, three_students):
    cid, sid = course["id"], three_students[0]
    client.post(f"/courses/{cid}/students/{sid}", headers=staff_headers)
    assert client.delete(f"/students/{sid}", headers=admin_headers).status_code == 204
    detail = client.get(f"/courses/{cid}", headers=staff_headers).json()
    assert detail["enrollments"] == []


# ---------- transaction behaviour, tested at the crud layer ----------


def test_failed_enrollment_changes_nothing(db: Session):
    course = crud.create_course(db, CourseCreate(code="DS-201", title="Data Science", seats=1))
    s1 = crud.create_student(db, StudentCreate(name="A One", email="a@x.com", age=17, city="Surat", marks=50))
    s2 = crud.create_student(db, StudentCreate(name="B Two", email="b@x.com", age=17, city="Surat", marks=50))

    crud.enroll_student(db, course.id, s1.id)

    with pytest.raises(crud.EnrollmentError):
        crud.enroll_student(db, course.id, s2.id)

    db.refresh(course)
    assert course.seats_taken == 1
    assert len(crud.list_student_courses(db, s2.id)) == 0

Run the whole suite:

uv run pytest

63 passed in 4.2s

The last test is the important one: after a failed enrollment, seats_taken is still 1 and the second student has no enrollment. Nothing from the failed transaction leaked.

Where StudyDesk is after this article

studydesk/
├── alembic/versions/
│   └── c3d4e5f6a7b8_add_courses_and_enrollments.py
├── tests/test_courses.py            ← 14 tests
└── src/studydesk/
    ├── __init__.py                  ← courses router
    ├── crud.py                      ← courses, enroll_student, unenroll_student, EnrollmentError
    ├── models.py                    ← Course, Enrollment, relationships, cascades
    ├── schemas.py                   ← CourseCreate, CourseOut, EnrollmentOut, CourseWithStudentsOut, StudentCourseOut
    └── routers/
        ├── courses.py               ← /courses, /courses/{id}, /courses/{id}/students/{sid}
        └── students.py              ← + /students/{id}/courses

Route	Who	Success	Errors
`GET /courses`	logged in	200	401
`POST /courses`	admin	201	401, 403, 409, 422
`GET /courses/{id}`	logged in	200	404
`POST /courses/{id}/students/{sid}`	logged in	201	404, 409 (full / duplicate)
`DELETE /courses/{id}/students/{sid}`	logged in	204	404
`GET /students/{id}/courses`	logged in	200	404

Concept Quiz

Answers are below. No coding needed.

Q1. What is a foreign key, and what does ondelete="CASCADE" make the database do?

Q2. Why is enrollments a real table with its own columns instead of a plain two-column link table?

Q3. What is the difference between ForeignKey(...) on a column and relationship(...) on a class?

Q4. What does back_populates do?

Q5. Explain the N+1 problem with GET /courses/1 and how selectinload fixes it.

Q6. What does a transaction guarantee, and which three operations does enroll_student put inside one?

Q7. Why is the course row locked with with_for_update() before reading seats_taken?

Q8. Why does enroll_student raise EnrollmentError instead of HTTPException?

Q9. What does the bare raise inside except Exception: db.rollback(); raise do?

Q10. In Part 9, deleting a student removed the enrollment but left seats_taken unchanged. Why, and how would you fix it?

Answers

A column holding the primary key of a row in another table, enforced by the database. ondelete="CASCADE" makes PostgreSQL delete the dependent rows automatically when the referenced row is deleted.
Because we store data about the link itself (enrolled_at). A link with its own fields is an association object and needs a full model.
ForeignKey is the database column and constraint. relationship is a Python attribute SQLAlchemy adds so you can navigate between objects (course.enrollments, enrollment.student); it is not a column.
It connects the two ends of a relationship so that setting one side updates the other in memory, keeping student.enrollments and enrollment.student consistent.
Lazy loading fetches the course (1 query), then its enrollments (1), then each student separately (N). selectinload fetches each relationship level in one WHERE id IN (...) query, so it is 3 queries regardless of N.
All operations succeed together or none do. Checking the seat, inserting the enrollment, and incrementing seats_taken are committed as one unit.
So the value read is current and cannot change until this transaction commits. A second transaction must wait for the lock, then sees the updated count and fails the check instead of overbooking.
crud.py is the data layer and must not know about HTTP. The router catches EnrollmentError and chooses 404 or 409.
It re-raises the exception that was just caught, unchanged, after the rollback has run. The caller still sees the original error.
delete_student relies on the cascade, which removes enrollment rows without running unenroll_student's decrement. Fix: in delete_student, loop over student.enrollments and decrement each course's seats_taken (locking each course) before deleting, inside the same transaction.

Epilogue

Two clerks click Enrol on the last seat. One gets a 201, the other gets a polite 409. Twenty students, twenty seats, every time.

Neha runs a load test: two hundred requests at once against GET /courses/1. The API answers, but slowly, and the CPU is idle most of the time. "Every request sits waiting for PostgreSQL to reply, and while it waits, that worker can do nothing else," she says. "Python has a way to wait without blocking."

In the next article: async SQLAlchemy and async endpoints — async def routes, an async engine with asyncpg, async sessions, and what actually changes when the API stops blocking on I/O.

Python 19 - Students and Courses: SQLAlchemy Relationships and Transactions

Where StudyDesk is now

Part 1: Two Clerks, One Seat

Part 2: Relationships in a Database

Foreign keys

The three shapes

Part 3: The Models

Reading the new parts

The migration

Part 4: Schemas

Lazy loading

Eager loading with `selectinload`

Part 6: Transactions

What a transaction is

Sessions already use transactions

`with_for_update()`

Part 7: CRUD for Courses and Enrollments

Reading `enroll_student`

Reading `unenroll_student`

Part 8: Routers

Courses router

Reading it

A student's courses

Wire the router

Part 9: Walking Through It

Part 10: Tests

Where StudyDesk is after this article

Concept Quiz

Answers

Epilogue

Comments

python

More from this blog

Python 21 - StudyDesk Complete: The Final Project Walkthrough

Python - 20 Going Async: Async SQLAlchemy and Async Endpoints

Python 18 - Proving It Works: pytest and API Testing

Python 17 - Changing the Database Safely: Alembic Migrations

Where StudyDesk is now

Part 1: Two Clerks, One Seat

Part 2: Relationships in a Database

Foreign keys

The three shapes

Part 3: The Models

Reading the new parts

The migration

Part 4: Schemas

Part 5: Loading Related Data — the N+1 Problem

Lazy loading

Eager loading with selectinload

Part 6: Transactions

What a transaction is

Sessions already use transactions

with_for_update()

Part 7: CRUD for Courses and Enrollments

Reading enroll_student

Reading unenroll_student

Part 8: Routers

Courses router

Reading it

A student's courses

Wire the router

Part 9: Walking Through It

Part 10: Tests

Where StudyDesk is after this article

Concept Quiz

Answers

Epilogue

Comments

python

More from this blog

Python 21 - StudyDesk Complete: The Final Project Walkthrough

Python - 20 Going Async: Async SQLAlchemy and Async Endpoints

Python 18 - Proving It Works: pytest and API Testing

Python 17 - Changing the Database Safely: Alembic Migrations

Eager loading with `selectinload`

`with_for_update()`

Reading `enroll_student`

Reading `unenroll_student`