Python 21 - StudyDesk Complete: The Final Project Walkthrough

The whole repository in its final form, one command to run it, and a map of which article built which piece.

Part 1: Fifteen Articles, One Project

Aman opens the folder he created in Article 1 with uv init --package studydesk. It had one file. It now has an authenticated, tested, containerised API with a database, migrations, a job queue, and async everything. Nothing was thrown away between articles; every piece was added to the same project.

This article is the reference. If you skipped ahead, cloned the repository, or want to check your own copy against the finished one, every file is here.

---

Part 2: The Map

#	Article	What it added to StudyDesk
1	First FastAPI App	uv package layout, app, /docs, uv run studydesk
2	JSON, Status Codes, Path Params	GET /students/{id}, HTTPException, Path() validation
3	Query Parameters	filter, sort, paginate on GET /students
4	Pydantic	Student schema, constraints, EmailStr, validators, computed_field
5	CRUD	POST, PUT (partial), DELETE with correct status codes
6	CORS	CORSMiddleware, allowed origins
7	PostgreSQL + SQLAlchemy	config.py, database.py, models.py, schemas.py, get_db, DbSession
8	Database Endpoints	crud.py, routers/students.py, response_model, APIRouter
9	Docker	Dockerfile, docker-compose.yml, HOST/PORT/RELOAD
10	Auth	User, hashing, JWT, /auth/*, CurrentUser, AdminUser
11	Background Work	BackgroundTasks, Job table, tasks.py, worker.py, /jobs/*
12	Alembic	alembic/, baseline + phone migration, migrate service
13	pytest	tests/, rollback fixtures, dependency override
14	Relationships + Transactions	Course, Enrollment, selectinload, with_for_update, /courses/*
15	Async	asyncpg, AsyncSession, async def everywhere, async tests
16	This article	README, final layout, one-command setup

---

Part 3: The Final Layout

studydesk/
├── .dockerignore
├── .env                     ← local secrets (not committed)
├── .env.example
├── .gitignore
├── .python-version
├── Dockerfile
├── README.md
├── alembic.ini
├── docker-compose.yml
├── pyproject.toml
├── uv.lock
├── alembic/
│   ├── env.py
│   ├── script.py.mako
│   └── versions/
│       ├── a1b2c3d4e5f6_baseline_students_users_jobs.py
│       ├── b7c8d9e0f1a2_add_phone_to_students.py
│       └── c3d4e5f6a7b8_add_courses_and_enrollments.py
├── src/
│   └── studydesk/
│       ├── __init__.py          ← app, middleware, routers, main()
│       ├── config.py            ← .env → settings
│       ├── database.py          ← async engine, session, Base, get_db, DbSession
│       ├── models.py            ← Student, Course, Enrollment, User, Job
│       ├── schemas.py           ← all Pydantic input/output shapes
│       ├── crud.py              ← all database operations
│       ├── dependencies.py      ← auth dependencies
│       ├── security.py          ← hashing, JWT
│       ├── tasks.py             ← background work
│       ├── worker.py            ← job queue worker
│       ├── seed.py              ← initial data + first admin
│       ├── data/students.json   ← seed data
│       └── routers/
│           ├── __init__.py
│           ├── auth.py
│           ├── students.py
│           ├── courses.py
│           └── jobs.py
└── tests/
    ├── conftest.py
    ├── test_auth.py
    ├── test_crud.py
    ├── test_students.py
    ├── test_jobs.py
    └── test_courses.py

---

Part 4: Running It From Zero

With Docker (recommended)

git clone <your-repo-url> studydesk
cd studydesk
cp .env.example .env            # fill in SECRET_KEY and ADMIN_PASSWORD
docker compose up --build -d
docker compose exec api python -m studydesk.seed

Open http://localhost:8000/docs. Log in via POST /auth/login with the admin credentials from .env, paste the token into Authorize, and use the API.

Locally

cp .env.example .env            # set DATABASE_URL to your local PostgreSQL
uv sync
uv run alembic upgrade head
uv run python -m studydesk.seed
uv run studydesk                # terminal 1
uv run studydesk-worker         # terminal 2
uv run pytest                   # needs TEST_DATABASE_URL and the studydesk_test database

---

Part 5: One Request, End to End

Before the files, the path a single request takes through the finished system. POST /courses/1/students/3 with a valid staff token:

1. Uvicorn receives the HTTP request on the event loop.
2. CORSMiddleware adds allow headers if the Origin is in ALLOWED_ORIGINS.
3. FastAPI matches courses.router → enroll, and resolves its dependencies:
   • course_id, student_id from the path, validated by Path(gt=0) (422 if bad)
   • db: DbSession → get_db() opens an AsyncSession
   • user: CurrentUser → HTTPBearer reads the token → get_current_user decodes the JWT, loads the user (401 if any step fails)
4. The route calls crud.enroll_student, which inside one transaction locks the course, checks the student, the duplicate, and the seats, inserts the enrollment, increments seats_taken, commits. A rule failure raises EnrollmentError, rolled back.
5. The route maps EnrollmentError to 404 or 409, or returns the Enrollment row.
6. response_model=EnrollmentOut turns the row into JSON, embedding StudentOut with its computed grade.
7. get_db() resumes after its yield and closes the session.
8. Uvicorn sends 201 Created.

Every article contributed a step.

---

Part 6: Configuration and Infrastructure Files

.env.example

# Database (asyncpg driver for the app; Alembic swaps to psycopg automatically)
DATABASE_URL=postgresql+asyncpg://USER:PASSWORD@localhost:5432/studydesk
TEST_DATABASE_URL=postgresql+asyncpg://USER:PASSWORD@localhost:5432/studydesk_test

# CORS: comma-separated origins allowed to call the API from a browser
ALLOWED_ORIGINS=http://127.0.0.1:5500,http://localhost:5500,http://127.0.0.1:3000,http://localhost:3000

# Server
HOST=127.0.0.1
PORT=8000
RELOAD=true

# Auth
SECRET_KEY=generate-with: python -c "import secrets; print(secrets.token_hex(32))"
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=60

# First admin, created by `python -m studydesk.seed`
ADMIN_EMAIL=admin@studydesk.com
ADMIN_PASSWORD=choose-a-strong-password

.gitignore

.venv/
.env
__pycache__/
*.pyc
.pytest_cache/
.ruff_cache/

.dockerignore

.venv/
.git/
.env
__pycache__/
*.pyc
.pytest_cache/
.ruff_cache/
tests/

pyproject.toml

[project]
name = "studydesk"
version = "1.0.0"
description = "Student records API for a coaching institute, built with FastAPI."
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "alembic>=1.13",
    "asyncpg>=0.29",
    "fastapi[standard]>=0.115.0",
    "psycopg[binary]>=3.2",
    "pwdlib[argon2]>=0.2",
    "pyjwt>=2.9",
    "python-dotenv>=1.0",
    "sqlalchemy>=2.0",
]

[project.scripts]
studydesk = "studydesk:main"
studydesk-worker = "studydesk.worker:main"

[build-system]
requires = ["uv_build>=0.8.0,<0.9.0"]
build-backend = "uv_build"

[dependency-groups]
dev = [
    "httpx>=0.27",
    "pytest>=8.0",
    "pytest-asyncio>=0.24",
    "ruff>=0.6",
]

[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"

[tool.ruff]
line-length = 110

[tool.ruff.lint]
select = ["E", "F", "I"]

(Exact version numbers will match whatever uv add pinned on your machine; uv.lock holds the precise ones.)

Dockerfile

FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim

WORKDIR /app

ENV UV_PROJECT_ENVIRONMENT=/app/.venv \
    UV_PYTHON_DOWNLOADS=never \
    UV_COMPILE_BYTECODE=1

COPY pyproject.toml uv.lock README.md ./
RUN uv sync --frozen --no-dev --no-install-project

COPY src ./src
COPY alembic.ini ./
COPY alembic ./alembic
RUN uv sync --frozen --no-dev

ENV PATH="/app/.venv/bin:$PATH"

EXPOSE 8000

CMD ["studydesk"]

docker-compose.yml

services:
  db:
    image: postgres:16
    container_name: studydesk-db
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: studydesk
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres -d studydesk"]
      interval: 5s
      timeout: 5s
      retries: 10

  migrate:
    build: .
    container_name: studydesk-migrate
    command: ["alembic", "upgrade", "head"]
    env_file:
      - .env
    environment:
      DATABASE_URL: postgresql+asyncpg://postgres:postgres@db:5432/studydesk
    depends_on:
      db:
        condition: service_healthy

  api:
    build: .
    container_name: studydesk-api
    env_file:
      - .env
    environment:
      DATABASE_URL: postgresql+asyncpg://postgres:postgres@db:5432/studydesk
      HOST: 0.0.0.0
      PORT: "8000"
      RELOAD: "false"
    ports:
      - "8000:8000"
    depends_on:
      db:
        condition: service_healthy
      migrate:
        condition: service_completed_successfully

  worker:
    build: .
    container_name: studydesk-worker
    command: ["studydesk-worker"]
    env_file:
      - .env
    environment:
      DATABASE_URL: postgresql+asyncpg://postgres:postgres@db:5432/studydesk
    depends_on:
      db:
        condition: service_healthy
      migrate:
        condition: service_completed_successfully

volumes:
  postgres_data:

alembic.ini — as generated by alembic init, with one line blanked:

sqlalchemy.url =

alembic/env.py

from logging.config import fileConfig

from alembic import context
from sqlalchemy import engine_from_config, pool

from studydesk.config import DATABASE_URL
from studydesk.database import Base
from studydesk import models  # noqa: F401  (registers every table on Base)

config = context.config
config.set_main_option("sqlalchemy.url", DATABASE_URL.replace("+asyncpg", "+psycopg"))

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata


def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
        compare_type=True,
    )
    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(connection=connection, target_metadata=target_metadata, compare_type=True)
        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

The three migration files in alembic/versions/ are exactly as generated and shown in Articles 12 and 14. Their revision IDs on your machine will differ; the chain is baseline → add phone → add courses and enrollments.

---

Part 7: The Application Package

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="1.0.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"])
async def home():
    return {"message": "Welcome to StudyDesk API"}


@app.get("/db-check", tags=["General"], summary="Check the database connection")
async def db_check(db: DbSession):
    total = await 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)

src/studydesk/config.py

import os
from pathlib import Path

from dotenv import load_dotenv

BASE_DIR = Path(__file__).resolve().parent.parent.parent  # project root
load_dotenv(BASE_DIR / ".env")

DATABASE_URL: str = os.environ["DATABASE_URL"]
TEST_DATABASE_URL: str | None = os.getenv("TEST_DATABASE_URL")

ALLOWED_ORIGINS: list[str] = [
    origin.strip() for origin in os.getenv("ALLOWED_ORIGINS", "").split(",") if origin.strip()
]

HOST: str = os.getenv("HOST", "127.0.0.1")
PORT: int = int(os.getenv("PORT", "8000"))
RELOAD: bool = os.getenv("RELOAD", "true").lower() == "true"

SECRET_KEY: str = os.environ["SECRET_KEY"]
ALGORITHM: str = os.getenv("ALGORITHM", "HS256")
ACCESS_TOKEN_EXPIRE_MINUTES: int = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "60"))

ADMIN_EMAIL: str | None = os.getenv("ADMIN_EMAIL")
ADMIN_PASSWORD: str | None = os.getenv("ADMIN_PASSWORD")

src/studydesk/database.py

from collections.abc import AsyncGenerator
from typing import Annotated

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

from .config import DATABASE_URL

engine = create_async_engine(DATABASE_URL, echo=False)

SessionLocal = async_sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)


class Base(DeclarativeBase):
    """Parent class for every SQLAlchemy model."""


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    """Open an async database session for one request, and always close it."""
    async with SessionLocal() as session:
        yield session


DbSession = Annotated[AsyncSession, Depends(get_db)]

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}>"

src/studydesk/schemas.py

from datetime import datetime
from typing import Literal

from pydantic import BaseModel, ConfigDict, EmailStr, Field, computed_field, field_validator


def clean_text(value: str) -> str:
    """Strip spaces, reject digits, convert to Title Case."""
    value = value.strip()
    if any(char.isdigit() for char in value):
        raise ValueError("must not contain digits")
    return value.title()


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


class StudentBase(BaseModel):
    name: str = Field(min_length=2, max_length=50, description="Full name", examples=["Priya Sharma"])
    email: EmailStr = Field(description="Email address", examples=["priya@example.com"])
    phone: str | None = Field(
        default=None,
        pattern=r"^\+?[0-9]{10,15}$",
        description="Phone number, digits only, optional + prefix",
        examples=["+919876543210"],
    )
    age: int = Field(ge=5, le=100, description="Age in years", examples=[17])
    city: str = Field(min_length=2, max_length=50, description="City", examples=["Ahmedabad"])
    marks: int = Field(ge=0, le=100, description="Marks out of 100", examples=[88])

    @field_validator("name", "city")
    @classmethod
    def validate_text(cls, value: str) -> str:
        return clean_text(value)


class StudentCreate(StudentBase):
    """Input schema for POST /students."""


class StudentUpdate(BaseModel):
    """Input schema for PUT /students/{id}. Every field is optional."""

    name: str | None = Field(default=None, min_length=2, max_length=50, examples=["Priya Sharma"])
    email: EmailStr | None = Field(default=None, examples=["priya@example.com"])
    phone: str | None = Field(default=None, pattern=r"^\+?[0-9]{10,15}$", examples=["+919876543210"])
    age: int | None = Field(default=None, ge=5, le=100, examples=[17])
    city: str | None = Field(default=None, min_length=2, max_length=50, examples=["Ahmedabad"])
    marks: int | None = Field(default=None, ge=0, le=100, examples=[88])

    @field_validator("name", "city")
    @classmethod
    def validate_text(cls, value: str | None) -> str | None:
        if value is None:
            return value
        return clean_text(value)


class StudentOut(StudentBase):
    model_config = ConfigDict(from_attributes=True)

    id: int = Field(gt=0, description="Unique student ID", examples=[1])

    @computed_field
    @property
    def grade(self) -> str:
        if self.marks >= 85:
            return "A"
        if self.marks >= 70:
            return "B"
        if self.marks >= 50:
            return "C"
        return "D"


class StudentListOut(BaseModel):
    total: int
    total_pages: int
    offset: int
    limit: int
    count: int
    students: list[StudentOut]


# ---------- users and auth ----------


class UserCreate(BaseModel):
    email: EmailStr = Field(examples=["ritu@studydesk.com"])
    password: str = Field(min_length=8, max_length=128, examples=["a-strong-password"])


class LoginRequest(BaseModel):
    email: EmailStr = Field(examples=["ritu@studydesk.com"])
    password: str = Field(min_length=1, max_length=128, examples=["a-strong-password"])


class UserOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: EmailStr
    role: Literal["staff", "admin"]
    is_active: bool
    created_at: datetime


class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"


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


class JobOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    kind: str
    status: Literal["pending", "running", "done", "failed"]
    payload: dict
    result: dict | None
    error: str | None
    attempts: int
    max_attempts: int
    created_at: datetime
    started_at: datetime | None
    finished_at: datetime | None


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


class CourseCreate(BaseModel):
    code: str = Field(min_length=2, max_length=20, pattern=r"^[A-Za-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):
    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):
    model_config = ConfigDict(from_attributes=True)

    id: int
    enrolled_at: datetime
    student: StudentOut


class CourseWithStudentsOut(CourseOut):
    enrollments: list[EnrollmentOut]


class StudentCourseOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    enrolled_at: datetime
    course: CourseOut

(CourseCreate.code now accepts lowercase in the pattern; crud.create_course upper-cases it, which the Article 14 tests already relied on.)

src/studydesk/security.py

from datetime import datetime, timedelta, timezone

import jwt
from pwdlib import PasswordHash

from .config import ACCESS_TOKEN_EXPIRE_MINUTES, ALGORITHM, SECRET_KEY

password_hash = PasswordHash.recommended()


def hash_password(password: str) -> str:
    return password_hash.hash(password)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    return password_hash.verify(plain_password, hashed_password)


def create_access_token(subject: str, role: str) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {"sub": subject, "role": role, "exp": expire}
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)


def decode_access_token(token: str) -> dict | None:
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except jwt.PyJWTError:
        return None

src/studydesk/dependencies.py, src/studydesk/crud.py, src/studydesk/tasks.py, src/studydesk/worker.py, src/studydesk/seed.py, and all four files in src/studydesk/routers/ are exactly as shown in Article 15, Parts 5–7. They are the final versions.

src/studydesk/routers/__init__.py is empty.

src/studydesk/data/students.json is the ten-student file from Article 4 (without grade).

---

Part 8: The Remaining Test Files

tests/conftest.py, tests/test_auth.py, and tests/test_crud.py are as shown in Article 15. Here are the other three in their async form.

tests/test_students.py

import pytest
from httpx import AsyncClient


@pytest.fixture(autouse=True)
def no_real_email(monkeypatch):
    sent: list[tuple[str, str]] = []

    def fake_send(to_email: str, student_name: str) -> None:
        sent.append((to_email, student_name))

    monkeypatch.setattr("studydesk.routers.students.send_welcome_email", fake_send)
    return sent


# ---------- authentication on the whole router ----------


async def test_list_requires_login(client: AsyncClient):
    assert (await client.get("/students")).status_code == 401


async def test_create_requires_login(client: AsyncClient, sample_student):
    assert (await client.post("/students", json=sample_student)).status_code == 401


# ---------- create ----------


async def test_create_student(client: AsyncClient, staff_headers, sample_student, no_real_email):
    response = await client.post("/students", json=sample_student, headers=staff_headers)
    assert response.status_code == 201
    body = response.json()
    assert body["id"] > 0
    assert body["name"] == "Priya Sharma"
    assert body["phone"] == "+919876543210"
    assert body["grade"] == "A"
    assert no_real_email == [("priya@example.com", "Priya Sharma")]


async def test_create_cleans_name_and_city(client: AsyncClient, staff_headers, sample_student):
    sample_student["name"] = "   priya sharma "
    sample_student["city"] = "ahmedabad"
    response = await client.post("/students", json=sample_student, headers=staff_headers)
    assert response.status_code == 201
    assert response.json()["name"] == "Priya Sharma"
    assert response.json()["city"] == "Ahmedabad"


async def test_create_duplicate_email_is_409(client: AsyncClient, staff_headers, sample_student):
    assert (await client.post("/students", json=sample_student, headers=staff_headers)).status_code == 201
    assert (await client.post("/students", json=sample_student, headers=staff_headers)).status_code == 409


@pytest.mark.parametrize(
    "field, bad_value",
    [
        ("email", "not-an-email"),
        ("age", 3),
        ("marks", 150),
        ("name", "P"),
        ("name", "Priya 123"),
        ("phone", "12345"),
    ],
)
async def test_create_invalid_field_is_422(client: AsyncClient, staff_headers, sample_student, field, bad_value):
    sample_student[field] = bad_value
    response = await client.post("/students", json=sample_student, headers=staff_headers)
    assert response.status_code == 422


# ---------- read ----------


async def test_get_student(client: AsyncClient, staff_headers, sample_student):
    created = (await client.post("/students", json=sample_student, headers=staff_headers)).json()
    response = await client.get(f"/students/{created['id']}", headers=staff_headers)
    assert response.status_code == 200
    assert response.json()["email"] == "priya@example.com"


async def test_get_missing_student_is_404(client: AsyncClient, staff_headers):
    assert (await client.get("/students/999", headers=staff_headers)).status_code == 404


async def test_get_invalid_id_is_422(client: AsyncClient, staff_headers):
    assert (await client.get("/students/abc", headers=staff_headers)).status_code == 422
    assert (await client.get("/students/0", headers=staff_headers)).status_code == 422


# ---------- list, filter, sort, paginate ----------


@pytest.fixture
async def five_students(client: AsyncClient, staff_headers, sample_student):
    data = [
        ("Priya Sharma", "priya@example.com", "Ahmedabad", 88),
        ("Rahul Verma", "rahul@example.com", "Surat", 72),
        ("Sneha Patel", "sneha@example.com", "Ahmedabad", 95),
        ("Arjun Mehta", "arjun@example.com", "Vadodara", 58),
        ("Dev Joshi", "dev@example.com", "Surat", 45),
    ]
    for name, email, city, marks in data:
        payload = dict(sample_student, name=name, email=email, city=city, marks=marks)
        assert (await client.post("/students", json=payload, headers=staff_headers)).status_code == 201


async def test_list_all(client: AsyncClient, staff_headers, five_students):
    response = await client.get("/students", headers=staff_headers)
    assert response.status_code == 200
    body = response.json()
    assert body["total"] == 5
    assert body["count"] == 5
    assert body["total_pages"] == 1


async def test_list_filter_by_city(client: AsyncClient, staff_headers, five_students):
    body = (await client.get("/students?city=surat", headers=staff_headers)).json()
    assert body["total"] == 2
    assert {s["name"] for s in body["students"]} == {"Rahul Verma", "Dev Joshi"}


async def test_list_filter_by_grade(client: AsyncClient, staff_headers, five_students):
    body = (await client.get("/students?grade=A", headers=staff_headers)).json()
    assert body["total"] == 2
    assert all(s["grade"] == "A" for s in body["students"])


async def test_list_sort_by_marks_desc(client: AsyncClient, staff_headers, five_students):
    body = (await client.get("/students?sort_by=marks&order=desc", headers=staff_headers)).json()
    assert [s["marks"] for s in body["students"]] == [95, 88, 72, 58, 45]


async def test_list_pagination(client: AsyncClient, staff_headers, five_students):
    page1 = (await client.get("/students?limit=2&offset=0", headers=staff_headers)).json()
    page3 = (await client.get("/students?limit=2&offset=4", headers=staff_headers)).json()
    assert page1["total"] == 5
    assert page1["total_pages"] == 3
    assert page1["count"] == 2
    assert page3["count"] == 1


async def test_list_bad_sort_field_is_422(client: AsyncClient, staff_headers):
    assert (await client.get("/students?sort_by=email", headers=staff_headers)).status_code == 422


# ---------- update ----------


async def test_update_partial(client: AsyncClient, staff_headers, sample_student):
    created = (await client.post("/students", json=sample_student, headers=staff_headers)).json()
    response = await client.put(f"/students/{created['id']}", json={"marks": 60}, headers=staff_headers)
    assert response.status_code == 200
    body = response.json()
    assert body["marks"] == 60
    assert body["grade"] == "C"
    assert body["name"] == "Priya Sharma"


async def test_update_empty_body_is_400(client: AsyncClient, staff_headers, sample_student):
    created = (await client.post("/students", json=sample_student, headers=staff_headers)).json()
    assert (await client.put(f"/students/{created['id']}", json={}, headers=staff_headers)).status_code == 400


async def test_update_to_taken_email_is_409(client: AsyncClient, staff_headers, sample_student):
    first = (await client.post("/students", json=sample_student, headers=staff_headers)).json()
    second = (
        await client.post("/students", json=dict(sample_student, email="other@example.com"), headers=staff_headers)
    ).json()
    response = await client.put(f"/students/{second['id']}", json={"email": first["email"]}, headers=staff_headers)
    assert response.status_code == 409


# ---------- delete and authorization ----------


async def test_delete_as_staff_is_403(client: AsyncClient, staff_headers, sample_student):
    created = (await client.post("/students", json=sample_student, headers=staff_headers)).json()
    assert (await client.delete(f"/students/{created['id']}", headers=staff_headers)).status_code == 403


async def test_delete_as_admin(client: AsyncClient, staff_headers, admin_headers, sample_student):
    created = (await client.post("/students", json=sample_student, headers=staff_headers)).json()
    assert (await client.delete(f"/students/{created['id']}", headers=admin_headers)).status_code == 204
    assert (await client.get(f"/students/{created['id']}", headers=admin_headers)).status_code == 404


async def test_delete_missing_is_404(client: AsyncClient, admin_headers):
    assert (await client.delete("/students/999", headers=admin_headers)).status_code == 404

tests/test_jobs.py

from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

from studydesk import crud
from studydesk.tasks import TASKS
from studydesk.worker import run_one


async def _no_sleep(seconds: float) -> None:
    return None


async def test_queue_report_requires_admin(client: AsyncClient, staff_headers):
    assert (await client.post("/jobs/grade-report", headers=staff_headers)).status_code == 403


async def test_queue_report_returns_202_pending(client: AsyncClient, admin_headers):
    response = await client.post("/jobs/grade-report?city=Surat", headers=admin_headers)
    assert response.status_code == 202
    body = response.json()
    assert body["status"] == "pending"
    assert body["payload"] == {"city": "Surat"}
    assert body["attempts"] == 0


async def test_get_job(client: AsyncClient, admin_headers):
    created = (await client.post("/jobs/grade-report", headers=admin_headers)).json()
    response = await client.get(f"/jobs/{created['id']}", headers=admin_headers)
    assert response.status_code == 200
    assert response.json()["id"] == created["id"]


async def test_get_missing_job_is_404(client: AsyncClient, admin_headers):
    assert (await client.get("/jobs/999", headers=admin_headers)).status_code == 404


async def test_worker_runs_grade_report(
    client: AsyncClient, admin_headers, staff_headers, sample_student, db: AsyncSession, monkeypatch
):
    monkeypatch.setattr("studydesk.tasks.asyncio.sleep", _no_sleep)
    monkeypatch.setattr("studydesk.routers.students.send_welcome_email", lambda *a: None)

    await client.post("/students", json=sample_student, headers=staff_headers)
    created = (await client.post("/jobs/grade-report", headers=admin_headers)).json()

    assert await run_one(db) is True

    job = await crud.get_job(db, created["id"])
    assert job.status == "done"
    assert job.attempts == 1
    assert job.result["total"] == 1
    assert job.result["by_grade"]["A"] == 1


async def test_worker_retries_then_fails(db: AsyncSession, monkeypatch):
    async def always_fails(db, payload):
        raise RuntimeError("boom")

    monkeypatch.setitem(TASKS, "grade_report", always_fails)

    job = await crud.create_job(db, kind="grade_report")

    for attempt in range(1, 4):
        assert await run_one(db) is True
        await db.refresh(job)
        assert job.attempts == attempt
        assert job.status == ("failed" if attempt == 3 else "pending")

    assert "boom" in job.error


async def test_worker_with_empty_queue_returns_false(db: AsyncSession):
    assert await run_one(db) is False

tests/test_courses.py

import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession

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
async def course(client: AsyncClient, admin_headers) -> dict:
    response = await 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
async def three_students(client: AsyncClient, staff_headers, sample_student) -> list[int]:
    ids = []
    for i in range(3):
        payload = dict(sample_student, email=f"s{i}@example.com")
        ids.append((await client.post("/students", json=payload, headers=staff_headers)).json()["id"])
    return ids


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


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


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


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


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


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


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

    detail = (await 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]


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


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


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


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


async def test_student_courses(client: AsyncClient, staff_headers, course, three_students):
    sid = three_students[0]
    await client.post(f"/courses/{course['id']}/students/{sid}", headers=staff_headers)
    response = await 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"]


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


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


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


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


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

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

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

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

Note the last course test now also asserts seats_taken == 0 after deleting an enrolled student, because delete_student frees seats since Article 15.

uv run pytest

63 passed in 4.6s

---

Part 9: README

README.md

# StudyDesk API

Student records API for a coaching institute. Built with FastAPI, SQLAlchemy (async), PostgreSQL, and Docker.

## Features

- Students: CRUD with filtering, sorting, pagination, and validation
- Courses and enrollments with seat limits enforced in a transaction
- JWT authentication with staff and admin roles
- Background welcome emails and a PostgreSQL-backed job queue with a worker
- Alembic migrations, pytest suite, Docker Compose setup

## Quick start (Docker)

    cp .env.example .env          # set SECRET_KEY and ADMIN_PASSWORD
    docker compose up --build -d
    docker compose exec api python -m studydesk.seed

Open http://localhost:8000/docs, log in with POST /auth/login, paste the token into Authorize.

## Local development

    cp .env.example .env          # point DATABASE_URL at your PostgreSQL
    uv sync
    uv run alembic upgrade head
    uv run python -m studydesk.seed
    uv run studydesk              # API on http://127.0.0.1:8000
    uv run studydesk-worker       # job worker (separate terminal)

## Tests

Create a database named `studydesk_test`, set TEST_DATABASE_URL in .env, then:

    uv run pytest

## Migrations

    uv run alembic revision --autogenerate -m "describe the change"
    uv run alembic upgrade head

## Layout

    src/studydesk/      application package (routers, crud, models, schemas, ...)
    alembic/            database migrations
    tests/              pytest suite

---

Part 10: The Complete Route Table

Method	Path	Who	Success	From article
GET	/	anyone	200	1
GET	/db-check	anyone	200	7
POST	/auth/register	anyone	201	10
POST	/auth/login	anyone	200	10
GET	/auth/me	logged in	200	10
GET	/students	logged in	200	3, 8
POST	/students	logged in	201	5, 8, 11
GET	/students/{id}	logged in	200	2, 8
PUT	/students/{id}	logged in	200	5, 8
DELETE	/students/{id}	admin	204	5, 8, 10
GET	/students/{id}/courses	logged in	200	14
GET	/courses	logged in	200	14
POST	/courses	admin	201	14
GET	/courses/{id}	logged in	200	14
POST	/courses/{id}/students/{sid}	logged in	201	14
DELETE	/courses/{id}/students/{sid}	logged in	204	14
POST	/jobs/grade-report	admin	202	11
GET	/jobs/{id}	admin	200	11

Seventeen routes, every one validated, typed, authenticated where it should be, and covered by a test.

---

Part 11: Where to Go From Here

StudyDesk is complete for the series, not for the world. If you keep building it, these are the natural next steps, each a small project on top of what exists:

• Refresh tokens — so staff are not logged out every hour (extends security.py and routers/auth.py).
• Rate limiting on /auth/login — a middleware or a dependency with a counter in the database.
• A real email provider — replace the body of send_welcome_email; nothing else changes.
• Celery or RQ — move TASKS into a proper queue when job volume demands it; the task functions are already in the right shape.
• CI — a GitHub Actions workflow that starts PostgreSQL, sets TEST_DATABASE_URL, and runs uv run pytest on every push.
• Deployment — the Docker image runs unchanged on Render, Fly.io, Railway, or any VM; add HTTPS in front with a reverse proxy.
• Observability — structured logging and request IDs in a middleware; a /metrics endpoint.

Every one of these plugs into a seam you already built: a dependency, a router, a task function, a Compose service.

---

Final Quiz

A short one, across the whole series. Answers below.

Q1. Which file would you open to change what a student response looks like?

Q2. Which file would you open to add a new database table, and which command follows?

Q3. A new route must be admin-only. What is the one parameter you add?

Q4. A route is slow because it waits on PostgreSQL. What two things must be true for async to help?

Q5. Two requests try to take the last seat at once. Which single line prevents overbooking?

Q6. Where do secrets live, and how do they reach the Docker container?

Q7. How do tests use a different database without any application code changing?

Q8. What is the correct startup order on a fresh machine, and why does Compose enforce it?

---

Answers

1. schemas.py — StudentOut (or StudentBase if the field is shared with input).
2. models.py, then uv run alembic revision --autogenerate -m "..." followed by uv run alembic upgrade head.
3. admin: AdminUser.
4. The route is async def, and every wait inside it (down to the database driver) is an await, with no blocking calls.
5. select(Course).where(Course.id == course_id).with_for_update() inside enroll_student; the second transaction waits for the lock, then sees the updated seat count.
6. In .env, which is never committed and never copied into the image (.dockerignore). Compose injects the values at run time via env_file and environment.
7. conftest.py sets app.dependency_overrides[get_db] to a function yielding a session on the test engine. Routes only ever asked for DbSession, so they receive the test session without knowing.
8. Database healthy → alembic upgrade head → API and worker. Compose encodes it with depends_on conditions so the app never starts against a missing or half-migrated schema.

---

Epilogue

Aman pushes the final commit. The repository has sixteen articles' worth of decisions in it, and each one can be found by reading the file it lives in.

Neha opens /docs one last time, scrolls through the seventeen routes, and closes the laptop.

"Three hundred students, one API, any number of clients," she says. "That was the job on day one. Everything after that was just doing it properly."