Skip to content

Project_Guide.md

Latest commit

 

History

History
1196 lines (898 loc) · 43.1 KB

Project_Guide.md

File metadata and controls

1196 lines (898 loc) · 43.1 KB
Raw

Project 2: Personal Song Library API

Time: 60–90 minutes (this one's bigger than Project 1 — take breaks) Stack: Python 3.9+, FastAPI, SQLAlchemy, SQLite, pytest Editor: VS Code What you'll build: A real REST API with full CRUD over a database of 25 songs, searchable by tags, plus a "random song for my mood" endpoint. Auto-generated API docs. Logging. Metrics. Tests. The works.

Why this project

In Project 1 you were a client — you called someone else's API. Now you're the kitchen. You build the waiter that other apps would talk to.

The shift is huge. As a client, you only think about: what do I ask for, what do I get back. As an API author, you think about everything else: data modeling, validation, error responses, status codes, documentation, persistence, concurrency. This is what your dad's whole team does at AWS, just at a billion times the scale.

By the end of this you'll have:

  • A working API running on your laptop with 8 endpoints
  • A real SQLite database with 25 seeded songs
  • Auto-generated interactive API docs (you'll love this — FastAPI does this for free)
  • Logging that traces every request
  • Metrics you can hit via an endpoint
  • Tests that hit your API in-process (no need to run a server to test it)
  • A clean GitHub repo

Step 0: Setup (5 min)

Open VS Code, create the project folder

mkdir song-library-api
cd song-library-api
code .

Set up a virtual environment

python3 -m venv venv

Activate:

  • Mac/Linux: source venv/bin/activate
  • Windows PowerShell: venv\Scripts\Activate.ps1
  • Windows cmd: venv\Scripts\activate.bat

Install packages

pip install fastapi uvicorn "sqlalchemy>=2.0" pydantic pytest httpx
pip freeze > requirements.txt

Quick tour of what we just installed:

  • fastapi — the framework. Modern, fast, type-safe, auto-generates docs.
  • uvicorn — the actual web server that runs FastAPI apps.
  • sqlalchemy — the most popular Python ORM (Object-Relational Mapper). Lets you talk to a database using Python objects instead of raw SQL.
  • pydantic — data validation. FastAPI uses it under the hood.
  • pytest — testing framework (same as Project 1).
  • httpx — HTTP client. FastAPI's TestClient uses it to call your API in tests.

Verify everything installed

python -c "import fastapi, sqlalchemy, pydantic; print('all good')"

If you see all good, you're set. If there's an error, fix it before continuing.

Create the file structure

In VS Code, build this layout:

song-library-api/
├── app/
│   ├── __init__.py        # empty file — marks the folder as a Python package
│   ├── main.py            # FastAPI app + route handlers
│   ├── database.py        # DB engine, session, base model
│   ├── models.py          # SQLAlchemy table definitions
│   ├── schemas.py         # Pydantic request/response models
│   └── metrics.py         # in-memory metrics
├── tests/
│   ├── __init__.py        # empty
│   └── test_api.py        # API tests
├── seed.py                # script to populate the DB with 25 songs
├── requirements.txt       # already created above
├── .gitignore
└── README.md

__init__.py files are empty files that tell Python "this folder is a package." You can create them in VS Code with right-click → New File.

Step 1: Break down the feature like a SWE (5 min)

Same drill as Project 1.

The vague request

"Build a personal song library where I can store songs with tags and search them by mood."

Make it concrete

Resources: songs. That's the noun. Songs have:

  • id (auto-assigned)
  • title (required)
  • artist (required)
  • album (optional)
  • year (optional)
  • tags (a list of 5 short labels like happy, kpop, focus)
  • created_at (when added)

Endpoints I need:

Method Path Purpose
GET /songs List all songs (optionally filter by tag)
GET /songs/{id} Get one song
POST /songs Add a new song
PUT /songs/{id} Update an existing song
DELETE /songs/{id} Delete a song
GET /songs/search Search by multiple tags
GET /songs/random Get a random song matching a mood
GET /tags List all tags currently in use
GET /metrics Get observability metrics

What can go wrong:

  • User asks for a song ID that doesn't exist → 404
  • User sends a song with no title → 422 (validation error, FastAPI does this for free)
  • User sends fewer than 1 or more than 10 tags → 422
  • User searches for a tag that no song has → empty list (not an error)
  • User tries to delete a song that doesn't exist → 404
  • Database is locked → 500

The tag system

Each song has a list of short string tags. Examples:

  • Moods: happy, sad, chill, melancholic, nostalgic, energetic
  • Contexts: focus, study, workout, gym, driving, party, dinner
  • Genres: kpop, pop, rock, indie, jazz, classical, hiphop, lofi

Tags are free-form strings — users can invent new ones. The API doesn't restrict the vocabulary, only the count (5 per song, give or take).

Step 2: Design the structure (5 min)

Picture this:

                   ┌──────────────────┐
   HTTP request ──►│   main.py        │  routes + handlers
                   │   (FastAPI app)  │
                   └────────┬─────────┘
                            │ uses
                ┌───────────┼───────────┐
                ▼           ▼           ▼
         ┌──────────┐ ┌──────────┐ ┌──────────┐
         │ schemas  │ │ database │ │ metrics  │
         │(Pydantic)│ │ + models │ │          │
         │          │ │(SQLAlchemy)│         │
         └──────────┘ └────┬─────┘ └──────────┘
                           │
                           ▼
                     ┌──────────┐
                     │songs.db  │  SQLite file on disk
                     └──────────┘

The pattern:

  • A request comes in to a route in main.py
  • The route uses schemas (Pydantic) to validate the input
  • It uses models (SQLAlchemy) to read/write the database
  • It records a metric
  • It returns a Pydantic schema as the response

This is FastAPI's standard architecture. Once you see it once, you see it in every FastAPI project.

Step 3: Define the database model (app/models.py) (5 min)

A "model" in SQLAlchemy is a Python class that maps to a database table. Each attribute is a column.

"""
app/models.py — SQLAlchemy ORM models.

These classes are how we read and write the database in Python terms.
A `Song` object IS a row in the `songs` table.
"""

from datetime import datetime
from sqlalchemy import Integer, String, JSON, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


# Every SQLAlchemy model inherits from a Base class. We define it once here.
class Base(DeclarativeBase):
    pass


class Song(Base):
    # __tablename__ tells SQLAlchemy what to call the actual table in the DB.
    __tablename__ = "songs"

    # PYTHON BASIC: class attributes. These become columns in the table.
    # Mapped[int] is a type hint — modern SQLAlchemy 2.0 style.
    id:         Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    title:      Mapped[str] = mapped_column(String(200), nullable=False)
    artist:     Mapped[str] = mapped_column(String(200), nullable=False)
    album:      Mapped[str] = mapped_column(String(200), nullable=True)
    year:       Mapped[int] = mapped_column(Integer, nullable=True)

    # JSON column — stores the list of tags as a JSON array.
    # SQLite supports this natively. In a real production system with a lot
    # of tag-based querying, you'd probably model tags as a separate table
    # with a many-to-many relationship. For our scale, JSON is simpler and fine.
    tags:       Mapped[list] = mapped_column(JSON, nullable=False, default=list)

    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    def __repr__(self) -> str:
        # PYTHON BASIC: __repr__ is what shows up when you print() the object.
        # Useful for debugging.
        return f"<Song id={self.id} title={self.title!r} artist={self.artist!r}>"

A few things to notice:

  • We never write SQL ourselves. SQLAlchemy generates it from these class definitions.
  • The type hints aren't just documentation — they actually control the column types.
  • nullable=False means the column is required.

Step 4: Define request/response shapes (app/schemas.py) (5 min)

SQLAlchemy models describe what's in the database. Pydantic schemas describe what's in API requests and responses. They look similar but they're different things.

Why both? Because what a user sends to your API isn't always exactly what you store in the DB. A user shouldn't send an id when creating a song (the DB picks one). A user shouldn't send created_at (the server sets it). Schemas let you control exactly what's allowed in and what goes out.

"""
app/schemas.py — Pydantic models for request and response bodies.

These describe the SHAPE OF DATA going in and out of the API.
They're separate from database models so we can validate, transform, and
control exactly what's exposed.
"""

from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field, ConfigDict


class SongBase(BaseModel):
    """Fields shared by create and update — the user-editable fields."""
    title:  str = Field(..., min_length=1, max_length=200)
    artist: str = Field(..., min_length=1, max_length=200)
    album:  Optional[str] = Field(None, max_length=200)
    year:   Optional[int] = Field(None, ge=1900, le=2100)

    # tags is a list of strings. We enforce length and per-tag character limits.
    # FastAPI will automatically return a 422 if these aren't met. Free validation!
    tags:   list[str] = Field(..., min_length=1, max_length=10)


class SongCreate(SongBase):
    """Schema for POST /songs — incoming data when creating a song."""
    pass


class SongUpdate(SongBase):
    """Schema for PUT /songs/{id} — incoming data when updating."""
    pass


class SongRead(SongBase):
    """Schema for responses — what we send back to the client."""
    id: int
    created_at: datetime

    # PYDANTIC v2 BASIC: this tells Pydantic to read from ORM objects
    # (i.e., SQLAlchemy models) by attribute, not just from dicts.
    model_config = ConfigDict(from_attributes=True)


class MetricsResponse(BaseModel):
    """Schema for GET /metrics."""
    requests_total: int
    requests_by_endpoint: dict[str, int]
    songs_in_db: int
    errors_total: int

Step 5: Set up the database (app/database.py) (5 min)

This file creates the database engine and the session machinery FastAPI will use.

"""
app/database.py — database engine and session setup.
"""

import logging
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

from app.models import Base

logger = logging.getLogger(__name__)

# Path to the SQLite database file. It'll be created at the project root.
# In a real app this would be in an env var, not hardcoded.
DATABASE_URL = "sqlite:///./songs.db"

# The engine is the lowest-level interface to the database.
# check_same_thread=False is a SQLite-specific quirk for FastAPI.
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})

# A SessionLocal is a factory that produces database sessions.
# A "session" is one unit of work — a conversation with the DB.
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


def init_db() -> None:
    """Create all tables defined on Base. Safe to call multiple times."""
    logger.info("Initializing database schema")
    Base.metadata.create_all(bind=engine)


def get_db():
    """
    FastAPI dependency that gives a route handler a DB session.
    
    The 'yield' makes this a generator — FastAPI calls it before the request,
    yields the session, then runs the finally block after the response is sent.
    This guarantees the session is always closed, even if the handler crashes.
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

That get_db() function is FastAPI's dependency injection in action. We'll use it in every route that needs database access. It's elegant and you'll see it everywhere.

Step 6: Set up metrics (app/metrics.py) (3 min)

"""
app/metrics.py — simple in-memory observability metrics.

In production you'd push these to Prometheus, CloudWatch, Datadog etc.
Here we just keep counters and expose them via a /metrics endpoint.

Four metrics, same as Project 1's budget:
  1. requests_total          — every request that hits any endpoint
  2. requests_by_endpoint    — per-endpoint counts (useful for traffic shaping)
  3. songs_in_db             — set on demand from the DB
  4. errors_total            — every 4xx/5xx response
"""

from collections import defaultdict
from threading import Lock

# A Lock prevents two simultaneous requests from corrupting our counters.
# This is a real concurrency concern even in a tiny API.
_lock = Lock()

_state = {
    "requests_total":  0,
    "errors_total":    0,
    "requests_by_endpoint": defaultdict(int),
}


def record_request(endpoint: str) -> None:
    with _lock:
        _state["requests_total"] += 1
        _state["requests_by_endpoint"][endpoint] += 1


def record_error() -> None:
    with _lock:
        _state["errors_total"] += 1


def snapshot() -> dict:
    """Return a copy of the current metrics state."""
    with _lock:
        return {
            "requests_total": _state["requests_total"],
            "errors_total": _state["errors_total"],
            "requests_by_endpoint": dict(_state["requests_by_endpoint"]),
        }

Step 7: Build the FastAPI app (app/main.py) (15 min)

This is the big one. Take your time. Read the comments. Run it before you finish writing it — start the server early to see things working.

"""
app/main.py — FastAPI application with all routes.

Endpoints:
  GET    /                  health check
  GET    /songs             list / filter songs
  GET    /songs/search      search by multiple tags (AND or OR)
  GET    /songs/random      random song matching a tag
  GET    /songs/{id}        get one song
  POST   /songs             create a song
  PUT    /songs/{id}        update a song
  DELETE /songs/{id}        delete a song
  GET    /tags              list all unique tags
  GET    /metrics           observability metrics
"""

import logging
import random
from typing import Optional

from fastapi import FastAPI, Depends, HTTPException, Query, Request, status
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session

from app import metrics
from app.database import engine, get_db, init_db
from app.models import Song
from app.schemas import SongCreate, SongUpdate, SongRead, MetricsResponse


# ─── Logging configuration ─────────────────────────────────────────────────
# Configure once, at app startup. Modules get loggers via logging.getLogger.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)


# ─── FastAPI app instance ──────────────────────────────────────────────────
app = FastAPI(
    title="Song Library API",
    description="A personal song library, searchable by mood and tags.",
    version="1.0.0",
)


# ─── Startup hook: create tables if they don't exist ───────────────────────
@app.on_event("startup")
def on_startup():
    init_db()
    logger.info("Application started")


# ─── Middleware: log every request and record metrics ──────────────────────
# Middleware runs around every request — perfect for cross-cutting concerns.
@app.middleware("http")
async def log_and_meter(request: Request, call_next):
    endpoint = f"{request.method} {request.url.path}"
    logger.info("→ %s", endpoint)
    metrics.record_request(endpoint)

    response = await call_next(request)

    if response.status_code >= 400:
        metrics.record_error()
        logger.warning("← %s [%d]", endpoint, response.status_code)
    else:
        logger.info("← %s [%d]", endpoint, response.status_code)

    return response


# ─── Health check ──────────────────────────────────────────────────────────
@app.get("/")
def root():
    return {"status": "ok", "service": "song-library-api"}


# ─── List songs (with optional tag filter) ─────────────────────────────────
@app.get("/songs", response_model=list[SongRead])
def list_songs(
    tag: Optional[str] = Query(None, description="Filter by a single tag"),
    db: Session = Depends(get_db),
):
    """List all songs. If `tag` is provided, only return songs with that tag."""
    query = db.query(Song)
    songs = query.all()

    if tag:
        # SQLite's JSON support varies — easiest to filter in Python.
        # PYTHON BASIC: list comprehension with a condition.
        songs = [s for s in songs if tag in s.tags]
        logger.info("Filtered %d songs with tag=%s", len(songs), tag)

    return songs


# ─── Search by multiple tags ───────────────────────────────────────────────
@app.get("/songs/search", response_model=list[SongRead])
def search_songs(
    tags: list[str] = Query(..., description="Tags to match (e.g. ?tags=happy&tags=kpop)"),
    match: str = Query("any", regex="^(any|all)$", description="any = OR match, all = AND match"),
    db: Session = Depends(get_db),
):
    """
    Search songs by tags. 
      match=any → song has AT LEAST ONE of the given tags
      match=all → song has ALL of the given tags
    """
    all_songs = db.query(Song).all()
    tag_set = set(tags)         # PYTHON BASIC: set for fast membership checks

    if match == "all":
        # The song's tags must be a superset of what we asked for
        results = [s for s in all_songs if tag_set.issubset(set(s.tags))]
    else:
        # Any overlap counts
        results = [s for s in all_songs if tag_set & set(s.tags)]

    logger.info("Search tags=%s match=%s → %d results", tags, match, len(results))
    return results


# ─── Random song by tag (the fun feature) ──────────────────────────────────
@app.get("/songs/random", response_model=SongRead)
def random_song(
    tag: Optional[str] = Query(None, description="Pick from songs with this tag"),
    db: Session = Depends(get_db),
):
    """Return one random song. If `tag` is set, only from songs with that tag."""
    songs = db.query(Song).all()
    if tag:
        songs = [s for s in songs if tag in s.tags]

    if not songs:
        raise HTTPException(status_code=404, detail=f"No songs found for tag '{tag}'")

    chosen = random.choice(songs)
    logger.info("Random pick: %s (tag=%s)", chosen.title, tag)
    return chosen


# ─── Get one song by ID ────────────────────────────────────────────────────
@app.get("/songs/{song_id}", response_model=SongRead)
def get_song(song_id: int, db: Session = Depends(get_db)):
    song = db.get(Song, song_id)
    if song is None:
        raise HTTPException(status_code=404, detail=f"Song {song_id} not found")
    return song


# ─── Create a song ─────────────────────────────────────────────────────────
@app.post("/songs", response_model=SongRead, status_code=status.HTTP_201_CREATED)
def create_song(payload: SongCreate, db: Session = Depends(get_db)):
    """Create a new song. Pydantic validates the payload before we get here."""
    # PYTHON BASIC: ** unpacks a dict into keyword args.
    # payload.model_dump() converts the Pydantic schema to a plain dict.
    song = Song(**payload.model_dump())
    db.add(song)
    db.commit()
    db.refresh(song)        # refresh so we get the auto-generated id and created_at
    logger.info("Created song id=%d title=%r", song.id, song.title)
    return song


# ─── Update a song ─────────────────────────────────────────────────────────
@app.put("/songs/{song_id}", response_model=SongRead)
def update_song(song_id: int, payload: SongUpdate, db: Session = Depends(get_db)):
    song = db.get(Song, song_id)
    if song is None:
        raise HTTPException(status_code=404, detail=f"Song {song_id} not found")

    # PYTHON BASIC: iterating over dict.items() to set attributes.
    for field, value in payload.model_dump().items():
        setattr(song, field, value)

    db.commit()
    db.refresh(song)
    logger.info("Updated song id=%d", song.id)
    return song


# ─── Delete a song ─────────────────────────────────────────────────────────
@app.delete("/songs/{song_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_song(song_id: int, db: Session = Depends(get_db)):
    song = db.get(Song, song_id)
    if song is None:
        raise HTTPException(status_code=404, detail=f"Song {song_id} not found")

    db.delete(song)
    db.commit()
    logger.info("Deleted song id=%d", song_id)
    # 204 No Content — convention for successful DELETE. No body.
    return None


# ─── List all unique tags in the library ───────────────────────────────────
@app.get("/tags", response_model=list[str])
def list_tags(db: Session = Depends(get_db)):
    """Return every unique tag that appears on any song, sorted."""
    songs = db.query(Song).all()
    # PYTHON BASIC: set comprehension to collect unique values.
    unique = {tag for song in songs for tag in song.tags}
    return sorted(unique)


# ─── Metrics endpoint ──────────────────────────────────────────────────────
@app.get("/metrics", response_model=MetricsResponse)
def get_metrics(db: Session = Depends(get_db)):
    snap = metrics.snapshot()
    snap["songs_in_db"] = db.query(Song).count()
    return snap


# ─── Global error handler ──────────────────────────────────────────────────
# Catch anything unexpected so we never leak a stack trace to the client.
@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception):
    metrics.record_error()
    logger.exception("Unhandled error on %s %s", request.method, request.url.path)
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"},
    )

Start the server (do this NOW, before going further)

uvicorn app.main:app --reload

--reload means the server restarts whenever you save a file. Critical for dev.

You should see:

INFO:     Uvicorn running on http://127.0.0.1:8000
INFO:     Application startup complete.

Now the magic moment. Open a browser to http://127.0.0.1:8000/docs.

You'll see a full interactive API explorer that FastAPI generated automatically from your code. You can try every endpoint right from the browser. This is one of the killer features of FastAPI — the documentation isn't a separate thing you write, it's a byproduct of writing the code correctly with types.

Try GET /songs — it'll return [] because the DB is empty. Let's fix that.

Step 8: Seed the database (seed.py) (5 min)

This script populates the DB with 25 real songs across diverse moods and genres.

"""
seed.py — populate the database with 25 starter songs.

Run with:  python seed.py

Idempotent: deletes existing songs first so you can re-run anytime.
"""

import logging
from app.database import SessionLocal, init_db
from app.models import Song

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


# 25 songs, each with exactly 5 tags. Diverse moods, genres, and contexts.
SONGS = [
    # K-pop
    {"title": "Dynamite",         "artist": "BTS",            "album": "BE",          "year": 2020,
     "tags": ["kpop", "happy", "energetic", "pop", "dance"]},
    {"title": "How You Like That","artist": "BLACKPINK",      "album": "THE ALBUM",   "year": 2020,
     "tags": ["kpop", "energetic", "party", "dance", "hiphop"]},
    {"title": "Super Shy",        "artist": "NewJeans",       "album": "Get Up",      "year": 2023,
     "tags": ["kpop", "happy", "chill", "pop", "summer"]},
    {"title": "Spring Day",       "artist": "BTS",            "album": "You Never Walk Alone", "year": 2017,
     "tags": ["kpop", "sad", "nostalgic", "ballad", "melancholic"]},
    {"title": "LOVE DIVE",        "artist": "IVE",            "album": "LOVE DIVE",   "year": 2022,
     "tags": ["kpop", "energetic", "pop", "dance", "party"]},

    # Pop / mainstream
    {"title": "Blinding Lights",  "artist": "The Weeknd",     "album": "After Hours", "year": 2020,
     "tags": ["pop", "energetic", "driving", "synthwave", "workout"]},
    {"title": "Watermelon Sugar", "artist": "Harry Styles",   "album": "Fine Line",   "year": 2019,
     "tags": ["pop", "happy", "summer", "chill", "romantic"]},
    {"title": "Levitating",       "artist": "Dua Lipa",       "album": "Future Nostalgia", "year": 2020,
     "tags": ["pop", "happy", "party", "dance", "workout"]},
    {"title": "good 4 u",         "artist": "Olivia Rodrigo", "album": "SOUR",        "year": 2021,
     "tags": ["pop", "energetic", "breakup", "rock", "angry"]},
    {"title": "As It Was",        "artist": "Harry Styles",   "album": "Harry's House", "year": 2022,
     "tags": ["pop", "nostalgic", "chill", "indie", "melancholic"]},
    {"title": "Anti-Hero",        "artist": "Taylor Swift",   "album": "Midnights",   "year": 2022,
     "tags": ["pop", "melancholic", "indie", "chill", "introspective"]},
    {"title": "Cruel Summer",     "artist": "Taylor Swift",   "album": "Lover",       "year": 2019,
     "tags": ["pop", "summer", "energetic", "party", "romantic"]},
    {"title": "Espresso",         "artist": "Sabrina Carpenter", "album": "Short n' Sweet", "year": 2024,
     "tags": ["pop", "happy", "summer", "party", "dance"]},

    # Focus / chill / instrumental
    {"title": "Weightless",       "artist": "Marconi Union",  "album": "Weightless",  "year": 2011,
     "tags": ["ambient", "focus", "chill", "study", "calm"]},
    {"title": "Clair de Lune",    "artist": "Claude Debussy", "album": "Suite Bergamasque", "year": 1905,
     "tags": ["classical", "chill", "focus", "melancholic", "study"]},
    {"title": "River Flows in You","artist": "Yiruma",        "album": "First Love",  "year": 2001,
     "tags": ["classical", "chill", "romantic", "focus", "instrumental"]},
    {"title": "Take Five",        "artist": "Dave Brubeck",   "album": "Time Out",    "year": 1959,
     "tags": ["jazz", "chill", "focus", "study", "instrumental"]},
    {"title": "Lofi Beats Vol 1", "artist": "Lofi Girl",      "album": "Study Mix",   "year": 2022,
     "tags": ["lofi", "focus", "chill", "study", "instrumental"]},

    # Workout / motivation
    {"title": "Eye of the Tiger", "artist": "Survivor",       "album": "Eye of the Tiger", "year": 1982,
     "tags": ["rock", "workout", "energetic", "gym", "motivation"]},
    {"title": "Stronger",         "artist": "Kanye West",     "album": "Graduation",  "year": 2007,
     "tags": ["hiphop", "workout", "energetic", "gym", "motivation"]},
    {"title": "Lose Yourself",    "artist": "Eminem",         "album": "8 Mile",      "year": 2002,
     "tags": ["hiphop", "workout", "energetic", "motivation", "gym"]},

    # Indie / alternative
    {"title": "Heat Waves",       "artist": "Glass Animals",  "album": "Dreamland",   "year": 2020,
     "tags": ["indie", "chill", "melancholic", "romantic", "driving"]},

    # Classic rock / jazz
    {"title": "Bohemian Rhapsody","artist": "Queen",          "album": "A Night at the Opera", "year": 1975,
     "tags": ["rock", "classic", "energetic", "dramatic", "nostalgic"]},
    {"title": "Fly Me to the Moon","artist": "Frank Sinatra", "album": "It Might as Well Be Swing", "year": 1964,
     "tags": ["jazz", "romantic", "chill", "classic", "dinner"]},
    {"title": "Hotel California", "artist": "Eagles",         "album": "Hotel California", "year": 1976,
     "tags": ["rock", "classic", "nostalgic", "driving", "melancholic"]},
]


def seed():
    init_db()
    db = SessionLocal()
    try:
        # Wipe existing data so the script is idempotent (safe to re-run).
        deleted = db.query(Song).delete()
        logger.info("Cleared %d existing songs", deleted)

        # PYTHON BASIC: building a list of model instances from dicts.
        songs = [Song(**data) for data in SONGS]
        db.add_all(songs)
        db.commit()

        logger.info("Seeded %d songs", len(songs))
    finally:
        db.close()


if __name__ == "__main__":
    seed()

Run it:

python seed.py

You should see Seeded 25 songs. A songs.db file appeared in your project root.

Now go back to http://127.0.0.1:8000/docs, hit GET /songs, click "Try it out", click "Execute". You'll see your 25 songs come back as JSON.

Try these next, in the docs:

  • GET /songs/random?tag=focus → gets you a focus song
  • GET /songs/search?tags=workout&tags=hiphop&match=all → workout AND hiphop songs
  • GET /tags → all unique tags
  • GET /metrics → look at your request counts

This is your API working. Take a second to appreciate it.

Step 9: Call your API from Python (3 min)

In the dad's analogy from the flight: now you have a kitchen. Let's walk in as a customer. Create a quick file called client_demo.py at the project root (don't commit it, it's just to play):

"""client_demo.py — demonstrate calling our own API from a Python script."""
import httpx

BASE = "http://127.0.0.1:8000"

# Get a random focus song
r = httpx.get(f"{BASE}/songs/random", params={"tag": "focus"})
print("Random focus song:", r.json())

# Add a new song
new = {
    "title":  "Bad Habit",
    "artist": "Steve Lacy",
    "album":  "Gemini Rights",
    "year":   2022,
    "tags":   ["rnb", "chill", "indie", "romantic", "mellow"],
}
r = httpx.post(f"{BASE}/songs", json=new)
print("Created:", r.status_code, r.json())

Run it (while the server is still running in another terminal):

python client_demo.py

You called your own API. You are now both the kitchen and the customer.

Step 10: Write tests (tests/test_api.py) (10 min)

Real tests for a real API. FastAPI's TestClient lets you call your endpoints in-process — no need for the server to be running.

"""
tests/test_api.py — API tests using FastAPI's TestClient.

Each test gets a fresh in-memory database so tests don't pollute each other.
This is a key SWE skill: ISOLATING tests so they're deterministic.
"""

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.main import app
from app.database import get_db
from app.models import Base, Song   # Song import registers the model with Base


# ─── Test fixture: a fresh in-memory DB for each test ───────────────────────
@pytest.fixture
def client():
    # Use an in-memory SQLite DB. Wiped every test. Lightning fast.
    # StaticPool forces all connections to share one DB — without it,
    # each connection gets its own empty in-memory DB and create_all
    # tables won't be visible to test sessions.
    test_engine = create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    TestSession = sessionmaker(bind=test_engine)
    Base.metadata.create_all(bind=test_engine)

    def override_get_db():
        db = TestSession()
        try:
            yield db
        finally:
            db.close()

    # Replace the production DB dependency with the test one.
    # This is FastAPI's dependency override — clean and powerful.
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()


# ─── Helpers ────────────────────────────────────────────────────────────────
def make_song(title="Test Song", tags=None):
    return {
        "title":  title,
        "artist": "Test Artist",
        "album":  "Test Album",
        "year":   2024,
        "tags":   tags or ["test", "happy", "chill", "demo", "pop"],
    }


# ─── Tests ──────────────────────────────────────────────────────────────────

def test_health_check(client):
    r = client.get("/")
    assert r.status_code == 200
    assert r.json()["status"] == "ok"


def test_list_songs_empty(client):
    r = client.get("/songs")
    assert r.status_code == 200
    assert r.json() == []


def test_create_song(client):
    r = client.post("/songs", json=make_song("Created Song"))
    assert r.status_code == 201
    body = r.json()
    assert body["title"] == "Created Song"
    assert "id" in body
    assert "created_at" in body


def test_create_song_validation_missing_title(client):
    bad = make_song()
    del bad["title"]
    r = client.post("/songs", json=bad)
    assert r.status_code == 422   # FastAPI's validation error code


def test_create_song_validation_no_tags(client):
    bad = make_song()
    bad["tags"] = []
    r = client.post("/songs", json=bad)
    assert r.status_code == 422


def test_get_song(client):
    created = client.post("/songs", json=make_song("Findable")).json()
    r = client.get(f"/songs/{created['id']}")
    assert r.status_code == 200
    assert r.json()["title"] == "Findable"


def test_get_song_404(client):
    r = client.get("/songs/9999")
    assert r.status_code == 404
    assert "not found" in r.json()["detail"].lower()


def test_update_song(client):
    created = client.post("/songs", json=make_song("Old")).json()
    updated = make_song("New")
    r = client.put(f"/songs/{created['id']}", json=updated)
    assert r.status_code == 200
    assert r.json()["title"] == "New"


def test_delete_song(client):
    created = client.post("/songs", json=make_song("To Delete")).json()
    r = client.delete(f"/songs/{created['id']}")
    assert r.status_code == 204

    # Confirm it's gone
    r = client.get(f"/songs/{created['id']}")
    assert r.status_code == 404


def test_filter_by_tag(client):
    client.post("/songs", json=make_song("A", tags=["kpop", "happy", "dance", "pop", "energetic"]))
    client.post("/songs", json=make_song("B", tags=["rock", "sad", "ballad", "slow", "melancholic"]))

    r = client.get("/songs", params={"tag": "kpop"})
    assert r.status_code == 200
    titles = [s["title"] for s in r.json()]
    assert "A" in titles
    assert "B" not in titles


def test_search_match_all(client):
    client.post("/songs", json=make_song("Both",  tags=["happy", "kpop", "dance", "pop", "fun"]))
    client.post("/songs", json=make_song("OnlyHappy", tags=["happy", "indie", "chill", "pop", "summer"]))

    r = client.get("/songs/search", params={"tags": ["happy", "kpop"], "match": "all"})
    titles = [s["title"] for s in r.json()]
    assert titles == ["Both"]


def test_search_match_any(client):
    client.post("/songs", json=make_song("Kpop", tags=["kpop", "happy", "dance", "pop", "fun"]))
    client.post("/songs", json=make_song("Rock", tags=["rock", "sad", "ballad", "slow", "mood"]))

    r = client.get("/songs/search", params={"tags": ["kpop", "rock"], "match": "any"})
    titles = sorted(s["title"] for s in r.json())
    assert titles == ["Kpop", "Rock"]


def test_random_song(client):
    client.post("/songs", json=make_song("OnlyFocus", tags=["focus", "lofi", "chill", "study", "instrumental"]))
    r = client.get("/songs/random", params={"tag": "focus"})
    assert r.status_code == 200
    assert "focus" in r.json()["tags"]


def test_random_song_no_match(client):
    r = client.get("/songs/random", params={"tag": "nonexistent"})
    assert r.status_code == 404


def test_list_tags(client):
    client.post("/songs", json=make_song("A", tags=["a", "b", "c", "d", "e"]))
    client.post("/songs", json=make_song("B", tags=["c", "d", "e", "f", "g"]))

    r = client.get("/tags")
    assert r.status_code == 200
    assert r.json() == ["a", "b", "c", "d", "e", "f", "g"]


def test_metrics_endpoint(client):
    client.get("/songs")
    client.get("/songs")
    client.get("/songs/9999")  # generates a 404

    r = client.get("/metrics")
    assert r.status_code == 200
    body = r.json()
    assert body["requests_total"] >= 3
    assert body["errors_total"] >= 1

Run them:

pytest -v

You should see 16 tests, all green. If any fail, read the output and fix before continuing.

Some things to notice in these tests:

  • Each test gets a fresh DB. No pollution between tests. This is the gold standard.
  • The dependency_overrides trick is FastAPI gold. It lets you swap out any dependency for tests.
  • StaticPool is required for in-memory SQLite. Without it, each connection gets its own empty database and create_all tables aren't visible to test sessions. File-based SQLite doesn't need this.
  • We test both the happy path AND the error path. A test suite that only checks happy paths is barely a test suite.

Step 11: Write the README (5 min)

# Song Library API

A personal song library REST API. Stores songs with tags, supports full CRUD,
and lets you search by mood. Built with FastAPI + SQLAlchemy + SQLite.

## What's in the box

- 9 REST endpoints with full request/response validation
- 25 pre-seeded songs spanning k-pop, pop, focus, workout, jazz, classical, and more
- Tag-based search (AND or OR)
- A "random song for this mood" endpoint
- Auto-generated interactive API docs at `/docs`
- Per-endpoint logging and 4 observability metrics
- 16 automated tests with isolated in-memory databases

## Setup

```bash
git clone https://github.com/<your-username>/song-library-api.git
cd song-library-api
python3 -m venv venv
source venv/bin/activate            # Windows: venv\Scripts\activate
pip install -r requirements.txt
python seed.py                       # populate the DB
```

## Run

```bash
uvicorn app.main:app --reload
```

The API is now at `http://127.0.0.1:8000`.

Interactive docs at `http://127.0.0.1:8000/docs`.

## Endpoints

| Method | Path | Description |
|--------|------|-------------|
| GET | `/` | Health check |
| GET | `/songs` | List songs (optional `?tag=` filter) |
| GET | `/songs/search?tags=x&tags=y&match=all` | Search by multiple tags |
| GET | `/songs/random?tag=focus` | Random song matching a tag |
| GET | `/songs/{id}` | Get one song |
| POST | `/songs` | Create a song |
| PUT | `/songs/{id}` | Update a song |
| DELETE | `/songs/{id}` | Delete a song |
| GET | `/tags` | List all unique tags |
| GET | `/metrics` | Observability metrics |

## Tests

```bash
pytest -v
```

16 tests covering happy paths, error paths, and validation rules.

## Architecture

```
app/
├── main.py        FastAPI app and route handlers
├── database.py    SQLAlchemy engine and session factory
├── models.py      ORM models (the `songs` table)
├── schemas.py     Pydantic request/response models
└── metrics.py     In-memory observability counters
```

**Why two model systems (SQLAlchemy + Pydantic)?**
SQLAlchemy models describe what's *in the database*. Pydantic schemas describe
what's *over the wire*. Keeping them separate lets us validate inputs, omit
fields from responses (e.g., never expose internal flags), and version the
external API without changing the DB.

**Tags as JSON:** tags are stored as a JSON array on each song. For our scale
this is simpler than a proper many-to-many table. At higher scale you'd want
a `tags` table and a `song_tags` join table for indexed tag queries.

## License

MIT

Step 12: Push to GitHub (5 min)

.gitignore

venv/
__pycache__/
*.pyc
.pytest_cache/
.DS_Store
.vscode/
songs.db
*.sqlite
*.sqlite3

Note we ignore songs.db — the database is a local artifact, not something to commit. Anyone cloning the repo runs python seed.py to recreate it.

Git workflow

git init
git add .
git commit -m "Initial commit: FastAPI song library with CRUD, search, tests"

Create the GitHub repo

  1. Go to github.com/new
  2. Name it song-library-api
  3. Don't initialize with anything
  4. Click Create
git branch -M main
git remote add origin https://github.com/<your-username>/song-library-api.git
git push -u origin main

What you just practiced

A real SWE-grade API workflow:

  1. Translating a vague request into concrete endpoints with a route table
  2. Separating concerns: routes, schemas, ORM, metrics each in their own file
  3. Two-tier modeling: SQLAlchemy for the database, Pydantic for the API
  4. FastAPI dependency injection for DB sessions
  5. Middleware for cross-cutting concerns (logging + metrics on every request)
  6. HTTP semantics: 201 for create, 204 for delete, 404 for missing, 422 for validation
  7. Auto-generated API documentation as a side effect of typed code
  8. In-process testing with TestClient and dependency overrides
  9. Test isolation with a fresh in-memory DB per test
  10. Idempotent seed scripts (re-runnable without breaking state)
  11. Clean repo hygiene: .gitignore for runtime artifacts, requirements pinned

Stretch goals (optional)

  • Add a played_count field on songs, and a POST /songs/{id}/play endpoint that increments it. New feature, no breaking changes — practice the workflow.
  • Add pagination to GET /songs: ?limit=10&offset=0. This is how real APIs avoid returning 100K rows.
  • Sort options: GET /songs?sort=year&order=desc.
  • Refactor tags into a separate table. This is the "real" relational way. Steep climb in SQLAlchemy but worth it.
  • Add a /songs/stats endpoint that returns counts per tag, per year, per artist. Aggregations are a different muscle than CRUD.