🏗️ Part 8 — Backend Capstone Project

from pydantic import BaseModel   # Base class for all data validation schemas

# Schema for incoming requests — what the client must send to create a user
class UserCreate(BaseModel):
    name: str      # Required; any string
    email: str     # Required; ideally unique in the database
    age: int       # Required; must be an integer
    # No Config class needed here — UserCreate is not read from database rows

# Schema for outgoing responses — what the API returns to the client
class UserResponse(BaseModel):
    id: int        # Assigned by the database when the row is inserted
    name: str
    email: str
    age: int

    class Config:
        from_attributes = True   # Required for response models that map from DB row objects

import psycopg2   # PostgreSQL driver for Python
import os         # Read environment variables (credentials, hostnames)

def get_connection():
    # All connection details come from environment variables set in docker-compose.yml
    # Never hardcode passwords in source code
    return psycopg2.connect(
        host=os.getenv("DB_HOST", "db"),          # "db" is the Docker service name
        database=os.getenv("DB_NAME", "appdb"),
        user=os.getenv("DB_USER", "postgres"),
        password=os.getenv("DB_PASSWORD", "password"),
    )

from app.db.connection import get_connection
from app.schemas.user import UserCreate

def create_user(user: UserCreate) -> dict:
    conn = get_connection()   # Open a connection to the database
    try:
        with conn.cursor() as cur:
            # Parameterised query — %s placeholders are filled safely by psycopg2
            # RETURNING id gives us back the auto-generated id
            cur.execute(
                "INSERT INTO users (name, email, age) VALUES (%s, %s, %s) RETURNING id",
                (user.name, user.email, user.age),
            )
            user_id = cur.fetchone()[0]   # Extract the new id from the result
            conn.commit()                 # Make the insert permanent
            return {"id": user_id, "name": user.name, "email": user.email, "age": user.age}
    finally:
        conn.close()   # Always close the connection

def get_users() -> list:
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT id, name, email, age FROM users ORDER BY id")
            rows = cur.fetchall()   # List of tuples: [(1, "Alice", ...), ...]
            # Convert each tuple to a dict for JSON serialisation
            return [{"id": r[0], "name": r[1], "email": r[2], "age": r[3]} for r in rows]
    finally:
        conn.close()

def find_user(user_id: int) -> dict | None:
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT id, name, email, age FROM users WHERE id = %s", (user_id,))
            row = cur.fetchone()   # Returns None if no row matches
            if row is None:
                return None        # Caller will raise HTTPException 404
            return {"id": row[0], "name": row[1], "email": row[2], "age": row[3]}
    finally:
        conn.close()

from fastapi import APIRouter, HTTPException
from app.schemas.user import UserCreate, UserResponse
from app.services import user_service

# All routes in this file share the /users prefix and "users" tag in /docs
router = APIRouter(prefix="/users", tags=["users"])

# POST /users — create a new user; returns 201 on success
@router.post("/", response_model=UserResponse, status_code=201)
def create_user(user: UserCreate):
    return user_service.create_user(user)

# GET /users — return all users as a list
@router.get("/", response_model=list[UserResponse])
def get_users():
    return user_service.get_users()

# GET /users/{user_id} — return one user by id
@router.get("/{user_id}", response_model=UserResponse)
def get_user(user_id: int):
    user = user_service.find_user(user_id)
    if user is None:
        # HTTPException tells FastAPI to return a JSON error response
        raise HTTPException(status_code=404, detail="User not found")
    return user

from fastapi import FastAPI                         # The main FastAPI class
from fastapi.middleware.cors import CORSMiddleware  # Allows cross-origin requests
from app.routes import users                        # Import the users router

app = FastAPI(title="User API")   # Title appears in /docs

# CORS middleware lets external clients (browsers, test scripts) call the API
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],       # Allow any origin in development
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(users.router)   # Register all /users routes

# Health check — a quick way to confirm the server is up and responding
@app.get("/")
def health_check():
    return {"status": "ok"}

import requests   # install with pip (Windows) or pip3 (Mac/Linux)

BASE_URL = "http://localhost:8000"

def test_health():
    """Verify the server is running and responding."""
    response = requests.get(f"{BASE_URL}/")
    assert response.status_code == 200, f"Expected 200, got {response.status_code}"
    assert response.json()["status"] == "ok"
    print("✅ Health check passed")

def test_create_user():
    """Create a user and verify the response."""
    # Use a unique email each run to avoid duplicate key errors on repeated runs
    # (the email column has a UNIQUE constraint in the database)
    import time
    unique_email = f"alice_{int(time.time())}@example.com"
    response = requests.post(
        f"{BASE_URL}/users",
        json={"name": "Alice", "email": unique_email, "age": 28}
    )
    assert response.status_code == 201, f"Expected 201, got {response.status_code}"
    data = response.json()
    assert "id" in data, "Response should contain an id"
    assert data["name"] == "Alice"
    print(f"✅ Create user passed — id={data['id']}")
    return data["id"]   # Return the id for use in later tests

def test_get_users():
    """Fetch all users and verify at least one is returned."""
    response = requests.get(f"{BASE_URL}/users")
    assert response.status_code == 200, f"Expected 200, got {response.status_code}"
    users = response.json()
    assert isinstance(users, list), "Response should be a list"
    assert len(users) > 0, "There should be at least one user"
    print(f"✅ Get users passed — {len(users)} user(s) found")

def test_get_user_not_found():
    """Request a non-existent user and verify 404 is returned."""
    response = requests.get(f"{BASE_URL}/users/99999")
    assert response.status_code == 404, f"Expected 404, got {response.status_code}"
    print("✅ Get unknown user returns 404 — passed")

def test_invalid_data():
    """Send invalid data and verify 422 is returned."""
    response = requests.post(
        f"{BASE_URL}/users",
        json={"name": "Bob", "email": "bob@example.com", "age": "not-a-number"}
    )
    assert response.status_code == 422, f"Expected 422, got {response.status_code}"
    print("✅ Invalid data returns 422 — passed")

# Run all tests when the script is executed directly
if __name__ == "__main__":
    try:
        test_health()
        test_create_user()
        test_get_users()
        test_get_user_not_found()
        test_invalid_data()
        print("\n🎉 All tests passed!")
    except AssertionError as e:
        print(f"\n❌ Test failed: {e}")
    except requests.exceptions.ConnectionError:
        print("\n❌ Could not connect to the server.")
        print("   Make sure the server is running: docker compose up --build")