Phase 5: APIs & Backend

pip install "fastapi[standard]"
# [standard] includes uvicorn (the server) and extra tools

from fastapi import FastAPI

app = FastAPI(
    title="Task Manager API",
    description="Manage your tasks via REST",
    version="1.0.0"
)

@app.get("/")
def root():
    return {"message": "Task Manager API is running"}

@app.get("/health")
def health_check():
    return {"status": "ok"}

# Run with: uvicorn main:app --reload
# --reload restarts the server on every file save (dev mode)

uvicorn main:app --reload
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process
INFO:     Application startup complete.

from fastapi import FastAPI, Path, Query, HTTPException
from typing  import Optional

app = FastAPI()

# ── Path parameter — {task_id} in the route ──
@app.get("/tasks/{task_id}")
def get_task(
    task_id: int = Path(..., ge=1, description="The task ID")
):
    """
    FastAPI automatically:
    - converts task_id from string to int
    - validates it's >= 1
    - documents it in /docs
    - returns 422 if it's not an int
    """
    task = db.get_task(task_id)
    if not task:
        # Raise HTTPException to return proper HTTP error responses
        raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
    return task


# ── Query parameters — function args with defaults ──
@app.get("/tasks")
def list_tasks(
    done:     Optional[bool] = Query(default=None, description="Filter by done status"),
    priority: Optional[int]  = Query(default=None, ge=1, le=5),
    limit:    int             = Query(default=20, le=100),
    offset:   int             = Query(default=0,  ge=0),
):
    """
    GET /tasks                   → all tasks, first 20
    GET /tasks?done=false        → only pending
    GET /tasks?priority=5        → only critical
    GET /tasks?limit=5&offset=10 → pagination
    """
    tasks = db.all_tasks()
    if done is not None:
        tasks = [t for t in tasks if t["done"] == done]
    if priority:
        tasks = [t for t in tasks if t["priority"] == priority]
    return tasks[offset : offset + limit]


# ── Multiple path parameters ──
@app.get("/users/{user_id}/tasks/{task_id}")
def get_user_task(user_id: int, task_id: int):
    return {"user_id": user_id, "task_id": task_id}


# ── Enum path parameter — only specific string values ──
from enum import Enum

class SortField(str, Enum):
    created  = "created"
    priority = "priority"
    title    = "title"

@app.get("/tasks/sorted/{field}")
def sorted_tasks(field: SortField):
    # FastAPI validates and documents the allowed values
    return {"sorted_by": field.value}

from pydantic  import BaseModel, Field
from typing    import Optional
from datetime  import date
from fastapi   import FastAPI

app = FastAPI()

# ── Separate schemas for create vs response ──
# This is the right pattern — don't reuse one model for everything

class TaskCreate(BaseModel):
    """What the client sends when creating a task."""
    title:    str           = Field(..., min_length=1, max_length=200)
    priority: int           = Field(default=2, ge=1, le=5)
    due:      Optional[date] = None
    tags:     list[str]     = []

class TaskUpdate(BaseModel):
    """What the client sends when updating — all fields optional (PATCH)."""
    title:    Optional[str]  = Field(default=None, min_length=1)
    priority: Optional[int]  = Field(default=None, ge=1, le=5)
    done:     Optional[bool] = None
    due:      Optional[date] = None

class TaskResponse(BaseModel):
    """What the server returns — includes server-generated fields."""
    id:       int
    title:    str
    priority: int
    done:     bool
    due:      Optional[date]
    tags:     list[str]
    created:  str           # server sets this — client doesn't send it


# ── Routes using the schemas ──
@app.post("/tasks", response_model=TaskResponse, status_code=201)
def create_task(task: TaskCreate):
    """
    FastAPI automatically:
    1. Parses the JSON body
    2. Validates it against TaskCreate
    3. Returns 422 if invalid
    4. Calls your function with a proper TaskCreate object
    5. Serializes the return value through TaskResponse
    """
    new_task = db.create(task)
    return new_task   # response_model filters to only TaskResponse fields

@app.patch("/tasks/{task_id}", response_model=TaskResponse)
def update_task(task_id: int, update: TaskUpdate):
    task = db.get(task_id)
    if not task:
        raise HTTPException(404, "Task not found")
    # model_dump(exclude_unset=True) — only the fields the client sent
    changes = update.model_dump(exclude_unset=True)
    return db.update(task_id, changes)

from fastapi           import Request
from fastapi.responses import JSONResponse

# Catch any unhandled exception — return JSON, not an HTML crash page
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    import logging
    logging.getLogger("api").error("Unhandled error", exc_info=exc)
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"}
        # Never expose exc details to clients in production
    )

# Custom exception class + handler
class NotFoundError(Exception):
    def __init__(self, resource: str, id: int):
        self.resource = resource
        self.id       = id

@app.exception_handler(NotFoundError)
async def not_found_handler(request: Request, exc: NotFoundError):
    return JSONResponse(
        status_code=404,
        content={"detail": f"{exc.resource} {exc.id} not found"}
    )

pip install sqlmodel
# SQLModel includes SQLAlchemy and Pydantic — no separate installs

from sqlmodel import create_engine, SQLModel, Session

# SQLite for development — zero setup, file-based
DATABASE_URL = "sqlite:///tasks.db"

# PostgreSQL for production:
# DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"

engine = create_engine(
    DATABASE_URL,
    echo=True,   # logs every SQL query (disable in production)
    connect_args={"check_same_thread": False}  # SQLite-only setting
)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)   # creates tables if they don't exist

def get_session():
    with Session(engine) as session:
        yield session   # FastAPI dependency injection uses this

from sqlmodel  import SQLModel, Field
from typing    import Optional
from datetime  import date, datetime

# ── The base fields shared across schemas ──
class TaskBase(SQLModel):
    title:    str           = Field(min_length=1, max_length=200)
    priority: int           = Field(default=2, ge=1, le=5)
    done:     bool          = False
    due:      Optional[date] = None

# ── The database table — table=True creates the SQL table ──
class Task(TaskBase, table=True):
    id:      Optional[int] = Field(default=None, primary_key=True)
    created: datetime       = Field(default_factory=datetime.utcnow)

# ── API schemas — inherit from TaskBase ──
class TaskCreate(TaskBase):
    pass   # exactly the base fields — no id, no created

class TaskUpdate(SQLModel):
    title:    Optional[str]  = None
    priority: Optional[int]  = None
    done:     Optional[bool] = None
    due:      Optional[date] = None

class TaskResponse(TaskBase):
    id:      int
    created: datetime

from fastapi  import FastAPI, Depends, HTTPException
from sqlmodel  import Session, select
from models    import Task, TaskCreate, TaskUpdate, TaskResponse
from database  import get_session, create_db_and_tables

app = FastAPI()

@app.on_event("startup")
def on_startup():
    create_db_and_tables()  # runs once on server start


# ── CREATE ──
@app.post("/tasks", response_model=TaskResponse, status_code=201)
def create_task(task_in: TaskCreate, session: Session = Depends(get_session)):
    task = Task.model_validate(task_in)  # create Task from TaskCreate
    session.add(task)
    session.commit()
    session.refresh(task)  # load server-generated id and created
    return task


# ── READ ALL ──
@app.get("/tasks", response_model=list[TaskResponse])
def list_tasks(
    done:   bool | None = None,
    offset: int          = 0,
    limit:  int          = 20,
    session: Session    = Depends(get_session)
):
    stmt = select(Task)
    if done is not None:
        stmt = stmt.where(Task.done == done)
    return session.exec(stmt.offset(offset).limit(limit)).all()


# ── READ ONE ──
@app.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(task_id: int, session: Session = Depends(get_session)):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(404, f"Task {task_id} not found")
    return task


# ── UPDATE ──
@app.patch("/tasks/{task_id}", response_model=TaskResponse)
def update_task(task_id: int, update: TaskUpdate, session: Session = Depends(get_session)):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(404, f"Task {task_id} not found")
    changes = update.model_dump(exclude_unset=True)
    task.sqlmodel_update(changes)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task


# ── DELETE ──
@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(task_id: int, session: Session = Depends(get_session)):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(404, f"Task {task_id} not found")
    session.delete(task)
    session.commit()
    # Return None — status 204 means "no content"

from fastapi           import Security, HTTPException, status
from fastapi.security  import APIKeyHeader
import os, secrets

# Reads the X-API-Key header from every request
api_key_scheme = APIKeyHeader(name="X-API-Key")

VALID_API_KEYS = {
    os.getenv("API_KEY_ADMIN"): "admin",
    os.getenv("API_KEY_READONLY"): "reader",
}

def get_current_role(api_key: str = Security(api_key_scheme)) -> str:
    role = VALID_API_KEYS.get(api_key)
    if not role:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or missing API key"
        )
    return role

# Protect an endpoint — inject the dependency
@app.delete("/tasks/{task_id}", status_code=204)
def delete_task(
    task_id: int,
    role: str       = Depends(get_current_role),  # auth happens here
    session: Session = Depends(get_session)
):
    if role != "admin":
        raise HTTPException(403, "Admin role required")
    # ... delete logic

# Generate a secure API key
def generate_api_key() -> str:
    return secrets.token_urlsafe(32)   # 32 bytes = 43 characters, URL-safe

pip install "python-jose[cryptography]" passlib[bcrypt]
# jose = JWT creation/validation  |  passlib = password hashing

from jose          import JWTError, jwt
from passlib.context import CryptContext
from fastapi        import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from datetime       import datetime, timedelta
from pydantic       import BaseModel
import os

SECRET_KEY   = os.getenv("JWT_SECRET_KEY")  # long random string in .env
ALGORITHM    = "HS256"
TOKEN_EXPIRE = 30   # minutes

pwd_ctx = CryptContext(schemes=["bcrypt"])
oauth2  = OAuth2PasswordBearer(tokenUrl="token")

class Token(BaseModel):
    access_token: str
    token_type:   str

def hash_password(plain: str) -> str:
    return pwd_ctx.hash(plain)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_ctx.verify(plain, hashed)

def create_access_token(data: dict) -> str:
    payload = data.copy()
    payload["exp"] = datetime.utcnow() + timedelta(minutes=TOKEN_EXPIRE)
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def get_current_user(token: str = Depends(oauth2)):
    credentials_err = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload  = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if not username: raise credentials_err
    except JWTError:
        raise credentials_err
    user = db.get_user_by_name(username)
    if not user: raise credentials_err
    return user

# ── Login endpoint — returns a JWT ──
@app.post("/token", response_model=Token)
def login(form: OAuth2PasswordRequestForm = Depends()):
    user = db.get_user_by_name(form.username)
    if not user or not verify_password(form.password, user.hashed_password):
        raise HTTPException(401, "Incorrect username or password")
    token = create_access_token({"sub": user.username})
    return {"access_token": token, "token_type": "bearer"}

# ── Protected endpoint ──
@app.get("/me")
def get_me(current_user = Depends(get_current_user)):
    return current_user

from fastapi import Depends, FastAPI, Query
from sqlmodel import Session

app = FastAPI()

# ── Dependency: database session ──
def get_db() -> Session:
    with Session(engine) as session:
        yield session    # yield = context manager — cleans up after request

# ── Dependency: pagination params ──
class Pagination:
    def __init__(self,
        offset: int = Query(0,  ge=0),
        limit:  int = Query(20, le=100)
    ):
        self.offset = offset
        self.limit  = limit

# ── Dependency: current user (from JWT) ──
def get_user(token: str = Depends(oauth2_scheme)):
    return decode_and_validate_token(token)

# ── Chained dependency: admin user ──
def require_admin(user = Depends(get_user)):
    if user.role != "admin":
        raise HTTPException(403, "Admin access required")
    return user   # Depends can chain — admin depends on get_user


# ── Endpoints use dependencies — clean and readable ──
@app.get("/tasks")
def list_tasks(
    pagination: Pagination = Depends(Pagination),  # pagination logic
    session:    Session    = Depends(get_db),       # db session
    user                    = Depends(get_user),     # authentication
):
    # The function body only cares about business logic
    stmt = select(Task).where(Task.user_id == user.id)
    return session.exec(stmt.offset(pagination.offset).limit(pagination.limit)).all()

@app.delete("/tasks/{id}", status_code=204)
def delete_task(
    id:      int,
    session: Session = Depends(get_db),
    admin              = Depends(require_admin),   # chained: must be admin
):
    # Only admins reach this code
    task = session.get(Task, id)
    session.delete(task)
    session.commit()

pip install httpx pytest pytest-asyncio

import pytest
from fastapi.testclient import TestClient
from sqlmodel            import create_engine, SQLModel, Session
from sqlmodel.pool        import StaticPool

from main     import app
from database import get_session

# In-memory SQLite — created fresh for each test
TEST_ENGINE = create_engine(
    "sqlite://",            # no file — in memory only
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,   # same connection for whole test
)

def get_test_session():
    with Session(TEST_ENGINE) as session:
        yield session

@pytest.fixture(autouse=True)
def fresh_db():
    """Recreate tables before every test — clean slate."""
    SQLModel.metadata.create_all(TEST_ENGINE)
    yield
    SQLModel.metadata.drop_all(TEST_ENGINE)

@pytest.fixture
def client():
    """TestClient with the real database swapped for test database."""
    app.dependency_overrides[get_session] = get_test_session
    yield TestClient(app)
    app.dependency_overrides.clear()

import pytest

class TestCreateTask:
    def test_create_task_success(self, client):
        response = client.post("/tasks", json={
            "title":    "Learn FastAPI",
            "priority": 3
        })
        assert response.status_code == 201
        data = response.json()
        assert data["title"]    == "Learn FastAPI"
        assert data["priority"] == 3
        assert data["done"]     == False
        assert "id"             in data   # server assigned an ID
        assert "created"        in data

    def test_create_task_empty_title(self, client):
        response = client.post("/tasks", json={"title": ""})
        assert response.status_code == 422   # validation failure

    def test_create_task_invalid_priority(self, client):
        response = client.post("/tasks", json={"title": "Test", "priority": 10})
        assert response.status_code == 422


class TestGetTask:
    def test_get_existing_task(self, client):
        # Create a task first
        created = client.post("/tasks", json={"title": "Test"}).json()
        task_id = created["id"]

        response = client.get(f"/tasks/{task_id}")
        assert response.status_code == 200
        assert response.json()["id"] == task_id

    def test_get_nonexistent_task(self, client):
        response = client.get("/tasks/99999")
        assert response.status_code == 404


class TestUpdateTask:
    def test_partial_update(self, client):
        task = client.post("/tasks", json={"title": "Original", "priority": 1}).json()

        response = client.patch(f"/tasks/{task['id']}", json={"done": True})
        assert response.status_code == 200
        updated = response.json()
        assert updated["done"]     == True
        assert updated["title"]    == "Original"   # unchanged
        assert updated["priority"] == 1            # unchanged


# Run all tests:
# pytest tests/ -v