Chapter 47: Build a REST API with FastAPI and a Database

This is where everything clicks together.

You know FastAPI from Chapter 41. You know SQLAlchemy from Chapter 31. You know pytest from Chapter 28. You know JWT auth from Chapter 41. Now you're going to combine all of that into one coherent, production-ready project.

We're building a Product Catalogue API — a service that lets clients create, read, update, and delete products, organized into categories, with user authentication. By the end, you'll have an API you could actually ship.

What We're Building

GET    /products            -> list all products (with filtering, pagination)
GET    /products/{id}       -> get one product
POST   /products            -> create a product  (auth required)
PUT    /products/{id}       -> update a product  (auth required)
DELETE /products/{id}       -> delete a product  (auth required)

GET    /categories          -> list all categories
POST   /categories          -> create a category (auth required)

POST   /auth/register       -> register a new user
POST   /auth/token          -> log in, get a JWT token

GET    /health              -> health check (public)

Project Structure

catalogue/
├── src/
│   └── catalogue/
│       ├── __init__.py
│       ├── main.py           <- FastAPI app
│       ├── config.py         <- settings
│       ├── database.py       <- SQLAlchemy engine + session
│       ├── models.py         <- ORM models
│       ├── schemas.py        <- Pydantic schemas
│       ├── dependencies.py   <- shared dependencies
│       ├── routers/
│       │   ├── __init__.py
│       │   ├── auth.py
│       │   ├── products.py
│       │   └── categories.py
│       └── services/
│           ├── __init__.py
│           ├── auth.py
│           └── products.py
├── tests/
│   ├── conftest.py
│   ├── test_auth.py
│   ├── test_products.py
│   └── test_categories.py
└── pyproject.toml

Step 1 — Configuration (`config.py`)

# src/catalogue/config.py
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    app_name:     str = "Product Catalogue API"
    app_version:  str = "1.0.0"
    debug:        bool = False

    database_url: str = "sqlite:///./catalogue.db"

    secret_key:   str = "change-this-secret-key-in-production"
    algorithm:    str = "HS256"
    token_expiry_hours: int = 24

    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")


@lru_cache
def get_settings() -> Settings:
    return Settings()

Step 2 — Database (`database.py`)

# src/catalogue/database.py
from sqlalchemy import create_engine, event
from sqlalchemy.orm import DeclarativeBase, sessionmaker, Session
from catalogue.config import get_settings

settings = get_settings()

engine = create_engine(
    settings.database_url,
    connect_args={"check_same_thread": False},   # SQLite only
    echo=settings.debug,
)

# Enable WAL mode for SQLite — better concurrent reads
if settings.database_url.startswith("sqlite"):
    @event.listens_for(engine, "connect")
    def set_wal_mode(dbapi_conn, _):
        dbapi_conn.execute("PRAGMA journal_mode=WAL")


SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


class Base(DeclarativeBase):
    pass


def get_db():
    """FastAPI dependency that provides a database session."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Step 3 — ORM Models (`models.py`)

# src/catalogue/models.py
from datetime import datetime
from sqlalchemy import (
    String, Text, Float, Integer, Boolean,
    ForeignKey, DateTime, func
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from catalogue.database import Base


class User(Base):
    __tablename__ = "users"

    id:           Mapped[int]      = mapped_column(primary_key=True)
    username:     Mapped[str]      = mapped_column(String(50), unique=True, index=True)
    email:        Mapped[str]      = mapped_column(String(200), unique=True, index=True)
    hashed_password: Mapped[str]  = mapped_column(String(200))
    is_active:    Mapped[bool]     = mapped_column(Boolean, default=True)
    created_at:   Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    products: Mapped[list["Product"]] = relationship(back_populates="owner")


class Category(Base):
    __tablename__ = "categories"

    id:          Mapped[int] = mapped_column(primary_key=True)
    name:        Mapped[str] = mapped_column(String(100), unique=True, index=True)
    description: Mapped[str | None] = mapped_column(Text, nullable=True)

    products: Mapped[list["Product"]] = relationship(back_populates="category")


class Product(Base):
    __tablename__ = "products"

    id:          Mapped[int]   = mapped_column(primary_key=True)
    name:        Mapped[str]   = mapped_column(String(200), index=True)
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    price:       Mapped[float] = mapped_column(Float)
    stock:       Mapped[int]   = mapped_column(Integer, default=0)
    is_active:   Mapped[bool]  = mapped_column(Boolean, default=True)
    created_at:  Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    updated_at:  Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), onupdate=func.now()
    )

    category_id: Mapped[int | None] = mapped_column(
        ForeignKey("categories.id"), nullable=True
    )
    owner_id:    Mapped[int]  = mapped_column(ForeignKey("users.id"))

    category: Mapped["Category | None"] = relationship(back_populates="products")
    owner:    Mapped["User"]             = relationship(back_populates="products")

Step 4 — Pydantic Schemas (`schemas.py`)

# src/catalogue/schemas.py
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field, field_validator, EmailStr


# ── User ──────────────────────────────────────────────────────────────────────

class UserCreate(BaseModel):
    username: str  = Field(..., min_length=3, max_length=50)
    email:    str  = Field(..., description="Valid email address")
    password: str  = Field(..., min_length=8)

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if not any(c.isupper() for c in v):
            raise ValueError("Password must contain at least one uppercase letter")
        if not any(c.isdigit() for c in v):
            raise ValueError("Password must contain at least one digit")
        return v

    @field_validator("email")
    @classmethod
    def email_lowercase(cls, v: str) -> str:
        if "@" not in v:
            raise ValueError("Invalid email address")
        return v.lower()


class UserOut(BaseModel):
    id:         int
    username:   str
    email:      str
    is_active:  bool
    created_at: datetime

    model_config = {"from_attributes": True}


class Token(BaseModel):
    access_token: str
    token_type:   str = "bearer"


# ── Category ──────────────────────────────────────────────────────────────────

class CategoryCreate(BaseModel):
    name:        str            = Field(..., min_length=1, max_length=100)
    description: Optional[str] = None


class CategoryOut(BaseModel):
    id:          int
    name:        str
    description: Optional[str]

    model_config = {"from_attributes": True}


# ── Product ───────────────────────────────────────────────────────────────────

class ProductCreate(BaseModel):
    name:        str            = Field(..., min_length=1, max_length=200)
    description: Optional[str] = None
    price:       float          = Field(..., gt=0, description="Must be positive")
    stock:       int            = Field(0, ge=0)
    category_id: Optional[int] = None


class ProductUpdate(BaseModel):
    name:        Optional[str]   = Field(None, min_length=1, max_length=200)
    description: Optional[str]   = None
    price:       Optional[float] = Field(None, gt=0)
    stock:       Optional[int]   = Field(None, ge=0)
    is_active:   Optional[bool]  = None
    category_id: Optional[int]   = None


class ProductOut(BaseModel):
    id:          int
    name:        str
    description: Optional[str]
    price:       float
    stock:       int
    is_active:   bool
    created_at:  datetime
    updated_at:  datetime
    category:    Optional[CategoryOut]
    owner:       UserOut

    model_config = {"from_attributes": True}


class ProductList(BaseModel):
    items:  list[ProductOut]
    total:  int
    page:   int
    pages:  int

Step 5 — Services

Auth service (`services/auth.py`)

# src/catalogue/services/auth.py
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from sqlalchemy.orm import Session
from catalogue.config import get_settings
from catalogue import models, schemas

settings   = get_settings()
pwd_ctx    = CryptContext(schemes=["bcrypt"], deprecated="auto")


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_token(data: dict) -> str:
    payload = data.copy()
    payload["exp"] = datetime.utcnow() + timedelta(hours=settings.token_expiry_hours)
    return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm)


def decode_token(token: str) -> Optional[str]:
    try:
        payload = jwt.decode(
            token, settings.secret_key, algorithms=[settings.algorithm]
        )
        return payload.get("sub")
    except JWTError:
        return None


def create_user(db: Session, data: schemas.UserCreate) -> models.User:
    user = models.User(
        username=data.username,
        email=data.email,
        hashed_password=hash_password(data.password),
    )
    db.add(user)
    db.commit()
    db.refresh(user)
    return user


def authenticate_user(db: Session, username: str, password: str) -> Optional[models.User]:
    user = db.query(models.User).filter(models.User.username == username).first()
    if not user or not verify_password(password, user.hashed_password):
        return None
    return user


def get_user_by_username(db: Session, username: str) -> Optional[models.User]:
    return db.query(models.User).filter(models.User.username == username).first()

Product service (`services/products.py`)

# src/catalogue/services/products.py
from typing import Optional
from sqlalchemy.orm import Session
from sqlalchemy import select, func
from catalogue import models, schemas


def get_products(
    db:          Session,
    page:        int = 1,
    per_page:    int = 20,
    category_id: Optional[int] = None,
    search:      Optional[str] = None,
    min_price:   Optional[float] = None,
    max_price:   Optional[float] = None,
    in_stock:    bool = False,
) -> schemas.ProductList:
    q = db.query(models.Product).filter(models.Product.is_active == True)

    if category_id:
        q = q.filter(models.Product.category_id == category_id)
    if search:
        q = q.filter(models.Product.name.ilike(f"%{search}%"))
    if min_price is not None:
        q = q.filter(models.Product.price >= min_price)
    if max_price is not None:
        q = q.filter(models.Product.price <= max_price)
    if in_stock:
        q = q.filter(models.Product.stock > 0)

    total  = q.count()
    pages  = (total + per_page - 1) // per_page
    items  = q.offset((page - 1) * per_page).limit(per_page).all()

    return schemas.ProductList(items=items, total=total, page=page, pages=pages)


def get_product(db: Session, product_id: int) -> Optional[models.Product]:
    return db.query(models.Product).filter(models.Product.id == product_id).first()


def create_product(
    db:    Session,
    data:  schemas.ProductCreate,
    owner: models.User,
) -> models.Product:
    product = models.Product(**data.model_dump(), owner_id=owner.id)
    db.add(product)
    db.commit()
    db.refresh(product)
    return product


def update_product(
    db:      Session,
    product: models.Product,
    data:    schemas.ProductUpdate,
) -> models.Product:
    update_data = data.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(product, field, value)
    db.commit()
    db.refresh(product)
    return product


def delete_product(db: Session, product: models.Product) -> None:
    db.delete(product)
    db.commit()


def get_categories(db: Session) -> list[models.Category]:
    return db.query(models.Category).all()


def create_category(db: Session, data: schemas.CategoryCreate) -> models.Category:
    category = models.Category(**data.model_dump())
    db.add(category)
    db.commit()
    db.refresh(category)
    return category

Step 6 — Dependencies (`dependencies.py`)

# src/catalogue/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from catalogue.database import get_db
from catalogue.services import auth as auth_service
from catalogue import models

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")


def get_current_user(
    token: str    = Depends(oauth2_scheme),
    db:    Session = Depends(get_db),
) -> models.User:
    username = auth_service.decode_token(token)
    if not username:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token",
            headers={"WWW-Authenticate": "Bearer"},
        )
    user = auth_service.get_user_by_username(db, username)
    if not user or not user.is_active:
        raise HTTPException(status_code=401, detail="User not found or inactive")
    return user

Step 7 — Routers

Auth router (`routers/auth.py`)

# src/catalogue/routers/auth.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from catalogue.database import get_db
from catalogue.services import auth as auth_service
from catalogue import schemas

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


@router.post("/register", response_model=schemas.UserOut, status_code=201)
def register(data: schemas.UserCreate, db: Session = Depends(get_db)):
    if auth_service.get_user_by_username(db, data.username):
        raise HTTPException(status_code=400, detail="Username already taken")
    return auth_service.create_user(db, data)


@router.post("/token", response_model=schemas.Token)
def login(form: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = auth_service.authenticate_user(db, form.username, form.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
        )
    token = auth_service.create_token({"sub": user.username})
    return schemas.Token(access_token=token)

Products router (`routers/products.py`)

# src/catalogue/routers/products.py
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from catalogue.database import get_db
from catalogue.dependencies import get_current_user
from catalogue.services import products as product_service
from catalogue import models, schemas

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


@router.get("", response_model=schemas.ProductList)
def list_products(
    page:        int            = Query(1, ge=1),
    per_page:    int            = Query(20, ge=1, le=100),
    category_id: Optional[int]  = None,
    search:      Optional[str]  = None,
    min_price:   Optional[float] = None,
    max_price:   Optional[float] = None,
    in_stock:    bool           = False,
    db:          Session        = Depends(get_db),
):
    return product_service.get_products(
        db, page, per_page, category_id, search, min_price, max_price, in_stock
    )


@router.get("/{product_id}", response_model=schemas.ProductOut)
def get_product(product_id: int, db: Session = Depends(get_db)):
    product = product_service.get_product(db, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return product


@router.post("", response_model=schemas.ProductOut, status_code=201)
def create_product(
    data:         schemas.ProductCreate,
    db:           Session      = Depends(get_db),
    current_user: models.User  = Depends(get_current_user),
):
    return product_service.create_product(db, data, current_user)


@router.put("/{product_id}", response_model=schemas.ProductOut)
def update_product(
    product_id:   int,
    data:         schemas.ProductUpdate,
    db:           Session     = Depends(get_db),
    current_user: models.User = Depends(get_current_user),
):
    product = product_service.get_product(db, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    if product.owner_id != current_user.id:
        raise HTTPException(status_code=403, detail="Not your product")
    return product_service.update_product(db, product, data)


@router.delete("/{product_id}", status_code=204)
def delete_product(
    product_id:   int,
    db:           Session     = Depends(get_db),
    current_user: models.User = Depends(get_current_user),
):
    product = product_service.get_product(db, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    if product.owner_id != current_user.id:
        raise HTTPException(status_code=403, detail="Not your product")
    product_service.delete_product(db, product)

Step 8 — Main App (`main.py`)

# src/catalogue/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from catalogue.config import get_settings
from catalogue.database import Base, engine
from catalogue.routers import auth, products, categories

settings = get_settings()


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: create all tables
    Base.metadata.create_all(bind=engine)
    yield
    # Shutdown: nothing needed for SQLite


app = FastAPI(
    title=settings.app_name,
    version=settings.app_version,
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(auth.router)
app.include_router(products.router)
app.include_router(categories.router)


@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"},
    )


@app.get("/health", tags=["Health"])
def health():
    return {"status": "ok", "version": settings.app_version}

Step 9 — Tests

`tests/conftest.py`

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from catalogue.main import app
from catalogue.database import get_db, Base

TEST_DB_URL = "sqlite:///./test.db"

engine_test   = create_engine(TEST_DB_URL, connect_args={"check_same_thread": False})
SessionTest   = sessionmaker(bind=engine_test, autocommit=False, autoflush=False)


@pytest.fixture(autouse=True)
def setup_db():
    Base.metadata.create_all(bind=engine_test)
    yield
    Base.metadata.drop_all(bind=engine_test)


@pytest.fixture
def db():
    session = SessionTest()
    try:
        yield session
    finally:
        session.close()


@pytest.fixture
def client(db):
    def override_get_db():
        try:
            yield db
        finally:
            pass
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()


@pytest.fixture
def auth_headers(client):
    client.post("/auth/register", json={
        "username": "testuser",
        "email":    "test@example.com",
        "password": "TestPass1",
    })
    response = client.post("/auth/token", data={
        "username": "testuser",
        "password": "TestPass1",
    })
    token = response.json()["access_token"]
    return {"Authorization": f"Bearer {token}"}

`tests/test_auth.py`

def test_register_success(client):
    r = client.post("/auth/register", json={
        "username": "alice",
        "email":    "alice@example.com",
        "password": "StrongPass1",
    })
    assert r.status_code == 201
    data = r.json()
    assert data["username"] == "alice"
    assert "hashed_password" not in data


def test_register_duplicate_username(client):
    payload = {"username": "bob", "email": "bob@x.com", "password": "Pass1234"}
    client.post("/auth/register", json=payload)
    r = client.post("/auth/register", json=payload)
    assert r.status_code == 400


def test_register_weak_password(client):
    r = client.post("/auth/register", json={
        "username": "carol",
        "email":    "carol@x.com",
        "password": "weak",
    })
    assert r.status_code == 422


def test_login_success(client):
    client.post("/auth/register", json={
        "username": "dave", "email": "d@x.com", "password": "Dave1234"
    })
    r = client.post("/auth/token", data={"username": "dave", "password": "Dave1234"})
    assert r.status_code == 200
    assert "access_token" in r.json()


def test_login_wrong_password(client):
    client.post("/auth/register", json={
        "username": "eve", "email": "e@x.com", "password": "Eve12345"
    })
    r = client.post("/auth/token", data={"username": "eve", "password": "wrong"})
    assert r.status_code == 401

`tests/test_products.py`

def test_create_product(client, auth_headers):
    r = client.post("/products", json={
        "name": "Widget", "price": 9.99, "stock": 100
    }, headers=auth_headers)
    assert r.status_code == 201
    data = r.json()
    assert data["name"]  == "Widget"
    assert data["price"] == 9.99


def test_create_product_unauthenticated(client):
    r = client.post("/products", json={"name": "Widget", "price": 9.99})
    assert r.status_code == 401


def test_list_products(client, auth_headers):
    for i in range(5):
        client.post("/products", json={"name": f"Product {i}", "price": float(i+1)},
                    headers=auth_headers)
    r = client.get("/products")
    assert r.status_code == 200
    data = r.json()
    assert data["total"] == 5
    assert len(data["items"]) == 5


def test_get_product_not_found(client):
    r = client.get("/products/9999")
    assert r.status_code == 404


def test_update_product(client, auth_headers):
    create_r = client.post("/products", json={"name": "Old", "price": 5.0},
                           headers=auth_headers)
    pid = create_r.json()["id"]

    r = client.put(f"/products/{pid}", json={"name": "New", "price": 10.0},
                   headers=auth_headers)
    assert r.status_code == 200
    assert r.json()["name"] == "New"


def test_delete_product(client, auth_headers):
    create_r = client.post("/products", json={"name": "ToDelete", "price": 1.0},
                           headers=auth_headers)
    pid = create_r.json()["id"]

    r = client.delete(f"/products/{pid}", headers=auth_headers)
    assert r.status_code == 204

    r = client.get(f"/products/{pid}")
    assert r.status_code == 404


def test_search_products(client, auth_headers):
    client.post("/products", json={"name": "Python Book", "price": 29.99},
                headers=auth_headers)
    client.post("/products", json={"name": "Java Book", "price": 24.99},
                headers=auth_headers)

    r = client.get("/products?search=python")
    assert r.status_code == 200
    assert r.json()["total"] == 1
    assert r.json()["items"][0]["name"] == "Python Book"


def test_filter_by_price(client, auth_headers):
    for price in [5.0, 15.0, 50.0, 100.0]:
        client.post("/products", json={"name": f"Item £{price}", "price": price},
                    headers=auth_headers)
    r = client.get("/products?min_price=10&max_price=60")
    assert r.status_code == 200
    assert r.json()["total"] == 2

Run all tests:

pip install pytest httpx
pytest tests/ -v --tb=short

Step 10 — Run It

pip install fastapi uvicorn sqlalchemy pydantic-settings python-jose passlib[bcrypt]

# Run the API
uvicorn catalogue.main:app --reload --app-dir src

# Visit the interactive docs
open http://127.0.0.1:8000/docs

The Swagger UI at /docs lets you:

Register a user (POST /auth/register)
Log in (POST /auth/token) — copy the token
Click "Authorize" -> paste the token
Create, list, update, and delete products — all from the browser

What You Built

A complete, production-ready REST API with:

SQLAlchemy ORM with relationships (User -> Product -> Category)
Pydantic schemas for validation, transformation, and documentation
JWT authentication — register, login, protected endpoints
Ownership checks — you can only edit your own products
Filtering and pagination — search by name, filter by price/category/stock
Service layer — business logic separated from route handlers
Full test suite — auth tests, product CRUD tests, filter tests
Auto-generated docs at /docs and /redoc

What's Next?

Chapter 48 covers Building a Web Scraper and Data Pipeline — an end-to-end project that scrapes data, cleans it with pandas, stores it in SQLite, visualises the results, and schedules the whole thing to run automatically.

Chapter 47: Build a REST API with FastAPI and a Database

What We're Building

Project Structure

Step 1 — Configuration (config.py)

Step 2 — Database (database.py)

Step 3 — ORM Models (models.py)

Step 4 — Pydantic Schemas (schemas.py)

Step 5 — Services

Auth service (services/auth.py)

Product service (services/products.py)

Step 6 — Dependencies (dependencies.py)