✅ W04 Lab Solutions - Building APIs with FastAPI

These are the model solutions for the 💻 W04 Lab. The complete files are available for download at the bottom of this page.

Download the Solution Files

Download models.py | Download main.py | Download test_main.py

You’ll also need the data files from the 🖥️ W04 Lecture page:

Download waitrose.jsonl

How to Run

On Nuvolos

1. Open the regular VS Code app (not the Chromium + Selenium version)
2. Activate the food environment: conda activate food
3. Make sure you have the dependencies: pip install fastapi uvicorn pydantic
4. Place models.py, main.py, and waitrose.jsonl in the same folder
5. Start the server:

uvicorn main:app --reload

6. Open /proxy/8000/docs in your browser (the full URL will appear in the terminal output)

On Your Local Machine

Same steps, but access the docs at http://localhost:8000/docs.

Running the Tests

pip install pytest httpx
pytest test_main.py -v

httpx is needed by FastAPI’s TestClient.

---

🎯 Section 1: The Pydantic Model (models.py)

from pydantic import BaseModel, Field


class WaitroseProduct(BaseModel):
    """A single product scraped from Waitrose."""

    name: str
    category: str
    url: str
    barcode: list[str] = Field(min_length=1)
    food_type: list[dict] | None = None

Key points:

• barcode: list[str] = Field(min_length=1) ensures every product has at least one barcode. Pydantic raises a ValidationError if you pass an empty list.
• food_type: list[dict] | None = None accepts the nested structure from the JSONL data without enforcing its internal shape. In a production system you’d define a nested Pydantic model for this, but for the lab exercise this is sufficient.

Testing the validation:

from models import WaitroseProduct

# Valid product
product = WaitroseProduct(
    name="Waitrose Essential Bread",
    category="Bakery",
    url="https://www.waitrose.com/...",
    barcode=["5000128931830"],
)
print(product)
# ✅ Works

# Empty barcode list → ValidationError
try:
    WaitroseProduct(
        name="Test", category="Bakery", url="https://...", barcode=[]
    )
except Exception as e:
    print(f"❌ Validation error: {e}")

# Missing required field → ValidationError
try:
    WaitroseProduct(category="Bakery", url="https://...", barcode=["123"])
except Exception as e:
    print(f"❌ Missing field: {e}")

---

🎯 Section 2: The FastAPI Application (main.py)

import os

import pandas as pd

from fastapi import FastAPI, HTTPException

from models import WaitroseProduct


# ---------- Nuvolos proxy detection ----------

def __is_running_on_nuvolos():
    """
    If we are running this script from Nuvolos Cloud,
    there will be an environment variable called HOSTNAME
    which starts with 'nv-'
    """
    hostname = os.getenv("HOSTNAME")
    return hostname is not None and hostname.startswith("nv-")


if __is_running_on_nuvolos():
    app = FastAPI(
        title="Waitrose Products API",
        root_path="/proxy/8000/",
    )
else:
    app = FastAPI(title="Waitrose Products API")


# ---------- Load data once at startup ----------

df = pd.read_json("waitrose.jsonl", lines=True)
products_data = df.to_dict(orient="records")

print(f"✅ Loaded {len(products_data)} products")


# ---------- Endpoints ----------

@app.get("/product/{barcode}")
def get_product(barcode: str) -> WaitroseProduct:
    matched = df[df["barcode"].apply(lambda codes: barcode in codes)]

    if matched.empty:
        raise HTTPException(status_code=404, detail="Product not found")

    product = matched.iloc[0].to_dict()
    return WaitroseProduct(**product)


@app.get(
    "/products",
    response_model=list[WaitroseProduct],
    summary="Get Waitrose products",
    description="Returns products, optionally filtered by category or name.",
)
def get_products(
    category: str | None = None,
    name_contains: str | None = None,
) -> list[WaitroseProduct]:
    results = products_data

    if category:
        results = [p for p in results if p.get("category") == category]

    if name_contains:
        query = name_contains.lower()
        results = [p for p in results if query in p.get("name", "").lower()]

    return [WaitroseProduct(**p) for p in results]

Key design decisions:

• Data loaded once at startup, not per request. The pd.read_json(...) call runs when uvicorn imports the module. Every request reads from the products_data list in memory.
• Two endpoint patterns. /product/{barcode} uses a path parameter for single-item lookup; /products uses query parameters for filtered browsing.
• Filtering is cumulative. Each if block narrows results further, so you can combine ?category=bakery&name_contains=bread in a single request.
• Name search is case-insensitive. Both the query and product name are lowercased before comparison.

---

🎯 Bonus: pytest Tests (test_main.py)

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def test_get_all_products():
    """GET /products returns a non-empty list with expected fields."""
    response = client.get("/products")
    assert response.status_code == 200
    products = response.json()
    assert len(products) > 0
    assert "name" in products[0]
    assert "category" in products[0]
    assert "barcode" in products[0]


def test_filter_by_category():
    """Category filter returns only matching products."""
    response = client.get("/products?category=bakery")
    assert response.status_code == 200
    products = response.json()
    for p in products:
        assert p["category"] == "bakery"


def test_invalid_category_returns_empty():
    """Non-existent category returns an empty list, not an error."""
    response = client.get("/products?category=NonexistentCategory")
    assert response.status_code == 200
    assert response.json() == []


def test_name_contains_filter():
    """Name search is case-insensitive and matches substrings."""
    response = client.get("/products?name_contains=bread")
    assert response.status_code == 200
    products = response.json()
    for p in products:
        assert "bread" in p["name"].lower()


def test_combined_filters():
    """Multiple filters narrow results cumulatively."""
    response = client.get("/products?category=bakery&name_contains=bread")
    assert response.status_code == 200
    products = response.json()
    for p in products:
        assert p["category"] == "bakery"
        assert "bread" in p["name"].lower()

What these tests cover:

• Basic endpoint works and returns expected structure
• Category filtering returns only matching products
• Non-existent category returns empty list (not 404)
• Name search is case-insensitive
• Filters combine correctly

Run with pytest test_main.py -v to see each test’s pass/fail status.