Module 2 · Software Engineering

📖 5 min read

overview#

Exercise 1 — Environment & Postgres on Docker CE
Exercise 2 — FastAPI + Async SQLAlchemy + CRUD for TODOs
Exercise 3 — Pagination, Filtering, Validation, and Error Handling
Exercise 3 — Database Migrations (Alembic), Concurrency & Transactions

Target duration: Each lab 60–120 minutes. Capstone 2–4 hours. Prereqs: Python 3.10+, Linux with Docker CE installed (no Desktop), basic SQL, REST.

Lab 1 — Setup PostgreSQL on Docker CE (not Desktop)#

Goal#

Run a Postgres instance in Docker Engine - Community and verify connectivity.

Tasks#

Install Docker CE (Engine) on Linux (Ubuntu/Debian recommended).
- Verify with:
```
docker --version
docker info | grep -E "Server|Engine|Community"
docker compose version   # plugin v2 preferred
```
- Acceptance proof: docker info shows Server: Docker Engine - Community. Any evidence of “Docker Desktop” = fail.

Start PostgreSQL via Docker Compose (or docker run)

Compose file (compose.yaml):

services:
  db:
    image: postgres:16-alpine
    container_name: todo_db
    environment:
      POSTGRES_USER: todo_user
      POSTGRES_PASSWORD: todo_pass
      POSTGRES_DB: todo_db
    ports:
      - "5432:5432"
    volumes:
      - pg_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U $$POSTGRES_USER -d $$POSTGRES_DB"]
      interval: 5s
      timeout: 3s
      retries: 10
volumes:
  pg_data:
    driver: local

If compose plugin isn’t available, use:

docker network create todo_net
docker run -d --name todo_db \
  --net todo_net \
  -e POSTGRES_USER=todo_user \
  -e POSTGRES_PASSWORD=todo_pass \
  -e POSTGRES_DB=todo_db \
  -p 5432:5432 \
  -v pg_data:/var/lib/postgresql/data \
  postgres:16-alpine

Verify database connectivity

From host or a psql client container:

docker exec -it todo_db psql -U todo_user -d todo_db -c "\dt"

Acceptance Criteria#

Evidence docker info shows Community (CE).
Postgres container healthy (docker ps, healthcheck passing).
Can run SELECT 1 against the DB.

Deliverables#

Screenshot/log of docker info indicating CE, container status, and a successful query.

Rubric (10 pts)#

Docker CE installed & verified (3)
Postgres container running & healthy (4)
Connectivity verified (3)

Lab 2 — FastAPI + Async SQLAlchemy + Basic TODO CRUD#

Goal#

Build an API-only TODO service using FastAPI + SQLAlchemy 2.x async (asyncpg) connected to the Dockerized Postgres.

Suggested Project Structure#

app/
  core/config.py
  db/session.py
  db/models.py
  db/repository.py
  api/routes/todos.py
  main.py
alembic/  (created in Lab 4)
compose.yaml (later adds app service)
Dockerfile

Tasks#

Dependencies

pip install fastapi uvicorn sqlalchemy asyncpg pydantic pydantic-settings

Async SQLAlchemy setup

# app/db/session.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase): pass

DATABASE_URL = "postgresql+asyncpg://todo_user:todo_pass@localhost:5432/todo_db"

engine = create_async_engine(DATABASE_URL, echo=False, pool_size=5, max_overflow=10)
async_session = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

Model

# app/db/models.py
from sqlalchemy import String, Boolean, DateTime, func, Integer
from sqlalchemy.orm import Mapped, mapped_column
from .session import Base

class Todo(Base):
  __tablename__ = "todos"
  id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
  title: Mapped[str] = mapped_column(String(200))
  description: Mapped[str | None] = mapped_column(String(1000), nullable=True)
  completed: Mapped[bool] = mapped_column(Boolean, default=False)
  created_at: Mapped[DateTime] = mapped_column(DateTime, server_default=func.now())
  updated_at: Mapped[DateTime] = mapped_column(DateTime, onupdate=func.now(), nullable=True)

Schema (Pydantic) & Routes

# app/api/routes/todos.py
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.db.session import async_session
from app.db.models import Todo

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

class TodoCreate(BaseModel):
  title: str = Field(min_length=1, max_length=200)
  description: str | None = Field(default=None, max_length=1000)

class TodoRead(BaseModel):
  id: int
  title: str
  description: str | None
  completed: bool

  class Config:
    from_attributes = True  # SQLAlchemy 2.x

async def get_session() -> AsyncSession:
  async with async_session() as session:
    yield session

@router.post("", response_model=TodoRead, status_code=status.HTTP_201_CREATED)
async def create_todo(payload: TodoCreate, session: AsyncSession = Depends(get_session)):
  todo = Todo(title=payload.title, description=payload.description)
  session.add(todo)
  await session.commit()
  await session.refresh(todo)
  return todo

@router.get("/{todo_id}", response_model=TodoRead)
async def get_one(todo_id: int, session: AsyncSession = Depends(get_session)):
  todo = await session.get(Todo, todo_id)
  if not todo:
    raise HTTPException(status_code=404, detail="Todo not found")
  return todo

@router.patch("/{todo_id}", response_model=TodoRead)
async def toggle_complete(todo_id: int, session: AsyncSession = Depends(get_session)):
  todo = await session.get(Todo, todo_id)
  if not todo:
    raise HTTPException(status_code=404, detail="Todo not found")
  todo.completed = not todo.completed
  await session.commit()
  await session.refresh(todo)
  return todo

@router.delete("/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_one(todo_id: int, session: AsyncSession = Depends(get_session)):
  todo = await session.get(Todo, todo_id)
  if not todo:
    return  # 204 is fine for idempotent deletes
  await session.delete(todo)
  await session.commit()
  return

FastAPI App

# app/main.py
from fastapi import FastAPI
from app.api.routes.todos import router as todo_router
from app.db.session import engine, Base

app = FastAPI(title="TODO API")

@app.on_event("startup")
async def on_startup():
  # For lab 2 only: create tables programmatically (later use Alembic)
  async with engine.begin() as conn:
    await conn.run_sync(Base.metadata.create_all)

app.include_router(todo_router)

Run

uvicorn app.main:app --reload
# Test in /docs (Swagger) and with curl
curl -X POST http://localhost:8000/todos -H "Content-Type: application/json" -d '{"title":"Read paper"}'

Acceptance Criteria#

API responds with 201 on create, returns JSON aligned with schema.
GET, PATCH toggle, DELETE behave as defined.
/docs loads correctly with generated OpenAPI.

Deliverables#

Source code.
A short README with run instructions and sample curl calls.

Rubric (20 pts)#

Async DB setup (5)
Model + schema correctness (5)
CRUD endpoints working (8)
README & API docs (2)

Lab 4 — Migrations (Alembic), Concurrency & Transactions#

Goal#

Use Alembic for schema management and add transaction-safe operations and indexes.

Note: Alembic typically runs with a sync driver (e.g., psycopg/psycopg2) even if the app uses asyncpg. Keep migrations separate from runtime DB access.

Tasks#

Add fields: due_date (nullable), priority (enum or int).
Initialize Alembic
```
pip install alembic psycopg[binary]  # psycopg3
alembic init alembic
```
- Set sqlalchemy.url in alembic.ini to sync URL: postgresql+psycopg://todo_user:todo_pass@localhost:5432/todo_db
Create revision & upgrade
- Autogenerate revision or handwrite upgrade()/downgrade().
- Run:
```
alembic upgrade head
```
Indexes
- Index on completed, created_at.
Transaction-safe toggle
- Use SELECT ... FOR UPDATE or session transactions to avoid race conditions when toggling completed in high concurrency.

Acceptance Criteria#

Alembic revision adds new columns/indexes.
Toggle endpoint guarantees consistency under concurrent requests (explain your approach).

Deliverables#

Alembic migration files.
Brief concurrency note in README.

Rubric (25 pts)#

Alembic set up & run (10)
Correct schema changes & indexes (8)
Concurrency-safe operation (7)