Project Structure & Layered Architecture

bookshelf/
├── pyproject.toml
├── app/
│   ├── __init__.py
│   ├── main.py            # creates the FastAPI app, includes the routers
│   ├── settings.py        # pydantic-settings configuration
│   ├── dependencies.py    # Depends() providers and Annotated aliases
│   ├── db.py              # data layer: SQLite adapters and Unit of Work
│   ├── models/            # Pydantic schemas shared by all layers
│   │   ├── __init__.py
│   │   └── book.py
│   ├── routers/           # web layer: one APIRouter per resource
│   │   ├── __init__.py
│   │   ├── admin.py
│   │   ├── books.py
│   │   └── reviews.py
│   └── services/          # service layer: use cases, errors, ports
│       ├── __init__.py
│       ├── books.py
│       ├── errors.py
│       └── ports.py
└── tests/
    ├── __init__.py
    ├── conftest.py
    ├── fakes.py
    ├── test_api.py
    └── test_service.py

from fastapi import APIRouter, status

from ..dependencies import BookServiceDep
from ..models import Book, BookCreate
from . import reviews

router = APIRouter(
    prefix="/books",
    tags=["books"],
    responses={404: {"description": "Book not found"}},
)
router.include_router(reviews.router)


@router.get("/")
def list_books(service: BookServiceDep) -> list[Book]:
    return service.list_books()


@router.post(
    "/",
    status_code=status.HTTP_201_CREATED,
    responses={409: {"description": "ISBN already exists"}},
)
def create_book(data: BookCreate, service: BookServiceDep) -> Book:
    return service.add_book(data)


@router.get("/{book_id}")
def read_book(book_id: int, service: BookServiceDep) -> Book:
    return service.get_book(book_id)

from fastapi import Depends, FastAPI, Request
from fastapi.responses import JSONResponse

from .dependencies import require_admin
from .routers import admin, books
from .services.errors import BookNotFound, DuplicateIsbn

app = FastAPI(title="Bookshelf API", version="0.1.0")

app.include_router(books.router, prefix="/api/v1")
app.include_router(
    admin.router,
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(require_admin)],
    responses={403: {"description": "Admin token required"}},
)


@app.exception_handler(BookNotFound)
def book_not_found_handler(request: Request, exc: BookNotFound) -> JSONResponse:
    return JSONResponse(status_code=404, content={"detail": str(exc)})


@app.exception_handler(DuplicateIsbn)
def duplicate_isbn_handler(request: Request, exc: DuplicateIsbn) -> JSONResponse:
    return JSONResponse(status_code=409, content={"detail": str(exc)})

from fastapi import APIRouter

from ..dependencies import BookServiceDep

# Imagine this module is shared between projects: it declares no prefix,
# tags or dependencies. The including application decides.
router = APIRouter()


@router.get("/stats")
def stats(service: BookServiceDep) -> dict[str, int]:
    return {"books": len(service.list_books())}

from fastapi import APIRouter, Depends, FastAPI
from fastapi.testclient import TestClient

order: list[str] = []


def app_dep() -> None:
    order.append("app")


def include_dep() -> None:
    order.append("include")


def router_dep() -> None:
    order.append("router")


def decorator_dep() -> None:
    order.append("decorator")


router = APIRouter(prefix="/books", dependencies=[Depends(router_dep)])
app = FastAPI(dependencies=[Depends(app_dep)])
app.include_router(router, prefix="/api/v1", dependencies=[Depends(include_dep)])
app.include_router(router, prefix="/api/latest")


@router.get("/ping", dependencies=[Depends(decorator_dep)])  # added after include_router
def ping() -> dict[str, str]:
    return {"ok": "yes"}


client = TestClient(app)
print(client.get("/api/v1/books/ping").json(), order)
print(client.get("/api/latest/books/ping").status_code)
print(sorted(app.openapi()["paths"]))

{'ok': 'yes'} ['app', 'include', 'router', 'decorator']
200
['/api/latest/books/ping', '/api/v1/books/ping']

from fastapi import APIRouter, status

from ..dependencies import BookServiceDep
from ..models import Review, ReviewCreate

router = APIRouter(prefix="/{book_id}/reviews", tags=["reviews"])


@router.get("/")
def list_reviews(book_id: int, service: BookServiceDep) -> list[Review]:
    return service.list_reviews(book_id)


@router.post("/", status_code=status.HTTP_201_CREATED)
def create_review(book_id: int, data: ReviewCreate, service: BookServiceDep) -> Review:
    return service.add_review(book_id, data)

GET  /api/v1/books/{book_id}/reviews/  tags=['books', 'reviews']  responses=200, 404, 422
POST /api/v1/books/{book_id}/reviews/  tags=['books', 'reviews']  responses=201, 404, 422
GET  /api/v1/books/                    tags=['books']             responses=200, 404
POST /api/v1/books/                    tags=['books']             responses=201, 404, 409, 422
GET  /api/v1/books/{book_id}           tags=['books']             responses=200, 404, 422
GET  /admin/stats                      tags=['admin']             responses=200, 403, 422

[project]
name = "bookshelf"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = ["fastapi[standard]>=0.142,<0.143"]

[tool.fastapi]
entrypoint = "app.main:app"

$ fastapi dev
 🐍 Using import string: app.main:app

curl -s -X POST http://127.0.0.1:37411/api/v1/books/ \
  -H 'Content-Type: application/json' \
  -d '{"title":"Dune","author":"Frank Herbert","isbn":"9780441013593"}'

{"title":"Dune","author":"Frank Herbert","isbn":"9780441013593","id":1}

from pydantic import BaseModel, Field


class BookCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    author: str = Field(min_length=1, max_length=100)
    isbn: str = Field(pattern=r"^\d{13}$")


class Book(BookCreate):
    id: int


class ReviewCreate(BaseModel):
    rating: int = Field(ge=1, le=5)
    text: str = Field(default="", max_length=2000)


class Review(ReviewCreate):
    id: int
    book_id: int

@router.get("/{book_id}")
def read_book(book_id: int, service: BookServiceDep) -> Book:
    return service.get_book(book_id)

class BookshelfError(Exception):
    """Base class for domain errors; knows nothing about HTTP."""


class BookNotFound(BookshelfError):
    def __init__(self, book_id: int) -> None:
        super().__init__(f"Book {book_id} not found")
        self.book_id = book_id


class DuplicateIsbn(BookshelfError):
    def __init__(self, isbn: str) -> None:
        super().__init__(f"A book with ISBN {isbn} already exists")
        self.isbn = isbn

from types import TracebackType
from typing import Protocol, Self

from ..models import Book, BookCreate, Review, ReviewCreate


class BookRepository(Protocol):
    def add(self, data: BookCreate) -> Book: ...
    def get(self, book_id: int) -> Book | None: ...
    def get_by_isbn(self, isbn: str) -> Book | None: ...
    def list(self) -> list[Book]: ...


class ReviewRepository(Protocol):
    def add(self, book_id: int, data: ReviewCreate) -> Review: ...
    def list_for_book(self, book_id: int) -> list[Review]: ...


class UnitOfWork(Protocol):
    books: BookRepository
    reviews: ReviewRepository

    def __enter__(self) -> Self: ...
    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None: ...
    def commit(self) -> None: ...
    def rollback(self) -> None: ...

from ..models import Book, BookCreate, Review, ReviewCreate
from .errors import BookNotFound, DuplicateIsbn
from .ports import UnitOfWork


class BookService:
    """Use cases. Depends on the UnitOfWork port, never on a database."""

    def __init__(self, uow: UnitOfWork) -> None:
        self.uow = uow

    def add_book(self, data: BookCreate) -> Book:
        with self.uow:
            if self.uow.books.get_by_isbn(data.isbn) is not None:
                raise DuplicateIsbn(data.isbn)
            book = self.uow.books.add(data)
            self.uow.commit()
            return book

    def get_book(self, book_id: int) -> Book:
        with self.uow:
            book = self.uow.books.get(book_id)
            if book is None:
                raise BookNotFound(book_id)
            return book

    def list_books(self) -> list[Book]:
        with self.uow:
            return self.uow.books.list()

    def add_review(self, book_id: int, data: ReviewCreate) -> Review:
        with self.uow:
            if self.uow.books.get(book_id) is None:
                raise BookNotFound(book_id)
            review = self.uow.reviews.add(book_id, data)
            self.uow.commit()
            return review

    def list_reviews(self, book_id: int) -> list[Review]:
        with self.uow:
            if self.uow.books.get(book_id) is None:
                raise BookNotFound(book_id)
            return self.uow.reviews.list_for_book(book_id)

import secrets
from typing import Annotated

from fastapi import Depends, Header, HTTPException, status

from .db import SqliteUnitOfWork
from .services.books import BookService
from .services.ports import UnitOfWork
from .settings import Settings, get_settings

SettingsDep = Annotated[Settings, Depends(get_settings)]


def get_unit_of_work(settings: SettingsDep) -> UnitOfWork:
    return SqliteUnitOfWork(settings.database_path)


UnitOfWorkDep = Annotated[UnitOfWork, Depends(get_unit_of_work)]


def get_book_service(uow: UnitOfWorkDep) -> BookService:
    return BookService(uow)


BookServiceDep = Annotated[BookService, Depends(get_book_service)]


def require_admin(
    settings: SettingsDep,
    x_admin_token: Annotated[str | None, Header()] = None,
) -> None:
    expected = settings.admin_token
    if (
        expected is None
        or x_admin_token is None
        or not secrets.compare_digest(
            x_admin_token.encode(), expected.get_secret_value().encode()
        )
    ):
        raise HTTPException(status.HTTP_403_FORBIDDEN, "Admin token required")

import sqlite3
from types import TracebackType
from typing import Self

from .models import Book, BookCreate, Review, ReviewCreate

SCHEMA = """
CREATE TABLE IF NOT EXISTS books (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    author TEXT NOT NULL,
    isbn TEXT NOT NULL UNIQUE
);
CREATE TABLE IF NOT EXISTS reviews (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    book_id INTEGER NOT NULL REFERENCES books(id),
    rating INTEGER NOT NULL,
    text TEXT NOT NULL
);
"""


def create_schema(path: str) -> None:
    with sqlite3.connect(path) as conn:
        conn.executescript(SCHEMA)


class SqliteBookRepository:
    def __init__(self, conn: sqlite3.Connection) -> None:
        self.conn = conn

    def add(self, data: BookCreate) -> Book:
        cur = self.conn.execute(
            "INSERT INTO books (title, author, isbn) VALUES (?, ?, ?)",
            (data.title, data.author, data.isbn),
        )
        return Book(id=cur.lastrowid, **data.model_dump())

    def _row(self, row: sqlite3.Row | None) -> Book | None:
        return None if row is None else Book(**dict(row))

    def get(self, book_id: int) -> Book | None:
        return self._row(
            self.conn.execute("SELECT * FROM books WHERE id = ?", (book_id,)).fetchone()
        )

    def get_by_isbn(self, isbn: str) -> Book | None:
        return self._row(
            self.conn.execute("SELECT * FROM books WHERE isbn = ?", (isbn,)).fetchone()
        )

    def list(self) -> list[Book]:
        rows = self.conn.execute("SELECT * FROM books ORDER BY id").fetchall()
        return [Book(**dict(row)) for row in rows]


class SqliteReviewRepository:
    def __init__(self, conn: sqlite3.Connection) -> None:
        self.conn = conn

    def add(self, book_id: int, data: ReviewCreate) -> Review:
        cur = self.conn.execute(
            "INSERT INTO reviews (book_id, rating, text) VALUES (?, ?, ?)",
            (book_id, data.rating, data.text),
        )
        return Review(id=cur.lastrowid, book_id=book_id, **data.model_dump())

    def list_for_book(self, book_id: int) -> list[Review]:
        rows = self.conn.execute(
            "SELECT * FROM reviews WHERE book_id = ? ORDER BY id", (book_id,)
        ).fetchall()
        return [Review(**dict(row)) for row in rows]


class SqliteUnitOfWork:
    """One connection and one transaction per use case."""

    def __init__(self, path: str) -> None:
        self.path = path

    def __enter__(self) -> Self:
        self.conn = sqlite3.connect(self.path)
        self.conn.row_factory = sqlite3.Row
        self.books = SqliteBookRepository(self.conn)
        self.reviews = SqliteReviewRepository(self.conn)
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        self.rollback()  # no-op after commit(); undoes uncommitted work
        self.conn.close()

    def commit(self) -> None:
        self.conn.commit()

    def rollback(self) -> None:
        self.conn.rollback()

201 {'title': 'Dune', 'author': 'Frank Herbert', 'isbn': '9780441013593', 'id': 1}
409 {'detail': 'A book with ISBN 9780441013593 already exists'}
201 {'rating': 5, 'text': 'Spice', 'id': 1, 'book_id': 1}
404 {'detail': 'Book 99 not found'}
403 {'detail': 'Admin token required'}
200 {'books': 1}

from types import TracebackType
from typing import Self

from app.models import Book, BookCreate, Review, ReviewCreate


class FakeBookRepository:
    def __init__(self) -> None:
        self.items: dict[int, Book] = {}

    def add(self, data: BookCreate) -> Book:
        book = Book(id=len(self.items) + 1, **data.model_dump())
        self.items[book.id] = book
        return book

    def get(self, book_id: int) -> Book | None:
        return self.items.get(book_id)

    def get_by_isbn(self, isbn: str) -> Book | None:
        return next((b for b in self.items.values() if b.isbn == isbn), None)

    def list(self) -> list[Book]:
        return list(self.items.values())


class FakeReviewRepository:
    def __init__(self) -> None:
        self.items: list[Review] = []

    def add(self, book_id: int, data: ReviewCreate) -> Review:
        review = Review(id=len(self.items) + 1, book_id=book_id, **data.model_dump())
        self.items.append(review)
        return review

    def list_for_book(self, book_id: int) -> list[Review]:
        return [r for r in self.items if r.book_id == book_id]


class FakeUnitOfWork:
    """Same port as SqliteUnitOfWork, state kept in memory."""

    def __init__(self) -> None:
        self.books = FakeBookRepository()
        self.reviews = FakeReviewRepository()
        self.committed = False

    def __enter__(self) -> Self:
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> None:
        pass

    def commit(self) -> None:
        self.committed = True

    def rollback(self) -> None:
        pass

from collections.abc import Iterator
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from app.db import SqliteUnitOfWork, create_schema
from app.dependencies import get_unit_of_work
from app.main import app
from app.services.ports import UnitOfWork

from .fakes import FakeUnitOfWork


@pytest.fixture
def fake_uow() -> FakeUnitOfWork:
    return FakeUnitOfWork()


@pytest.fixture(params=["fake", "sqlite"])
def uow(request: pytest.FixtureRequest, tmp_path: Path) -> UnitOfWork:
    """The same tests run against every adapter of the UnitOfWork port."""
    if request.param == "fake":
        return FakeUnitOfWork()
    path = str(tmp_path / "test.db")
    create_schema(path)
    return SqliteUnitOfWork(path)


@pytest.fixture
def client(fake_uow: FakeUnitOfWork) -> Iterator[TestClient]:
    app.dependency_overrides[get_unit_of_work] = lambda: fake_uow
    with TestClient(app) as test_client:
        yield test_client
    app.dependency_overrides.clear()

import pytest

from app.models import BookCreate, ReviewCreate
from app.services.books import BookService
from app.services.errors import BookNotFound, DuplicateIsbn
from app.services.ports import UnitOfWork

DUNE = BookCreate(title="Dune", author="Frank Herbert", isbn="9780441013593")


def test_add_and_get(uow: UnitOfWork) -> None:
    service = BookService(uow)
    created = service.add_book(DUNE)
    assert service.get_book(created.id) == created


def test_duplicate_isbn_rejected(uow: UnitOfWork) -> None:
    service = BookService(uow)
    service.add_book(DUNE)
    with pytest.raises(DuplicateIsbn):
        service.add_book(DUNE)


def test_review_for_missing_book(uow: UnitOfWork) -> None:
    with pytest.raises(BookNotFound):
        BookService(uow).add_review(99, ReviewCreate(rating=5))


def test_reviews_are_listed_per_book(uow: UnitOfWork) -> None:
    service = BookService(uow)
    book = service.add_book(DUNE)
    service.add_review(book.id, ReviewCreate(rating=5, text="Spice"))
    assert [r.text for r in service.list_reviews(book.id)] == ["Spice"]

from unittest.mock import create_autospec

from fastapi.testclient import TestClient

from app.dependencies import get_book_service
from app.main import app
from app.services.books import BookService
from app.services.errors import BookNotFound

from .fakes import FakeUnitOfWork

BOOK = {"title": "Dune", "author": "Frank Herbert", "isbn": "9780441013593"}


def test_create_and_read_with_a_fake(client: TestClient, fake_uow: FakeUnitOfWork) -> None:
    response = client.post("/api/v1/books/", json=BOOK)
    assert response.status_code == 201
    assert fake_uow.committed
    assert client.get("/api/v1/books/1").json()["title"] == "Dune"


def test_duplicate_isbn_is_409(client: TestClient) -> None:
    client.post("/api/v1/books/", json=BOOK)
    assert client.post("/api/v1/books/", json=BOOK).status_code == 409


def test_nested_review_route(client: TestClient) -> None:
    client.post("/api/v1/books/", json=BOOK)
    created = client.post("/api/v1/books/1/reviews/", json={"rating": 5})
    assert created.status_code == 201
    assert client.get("/api/v1/books/1/reviews/").json()[0]["book_id"] == 1


def test_web_layer_alone_with_a_mock(client: TestClient) -> None:
    """A mock checks the interaction: the router must call the service."""
    service = create_autospec(BookService, instance=True)
    service.get_book.side_effect = BookNotFound(7)
    app.dependency_overrides[get_book_service] = lambda: service
    response = client.get("/api/v1/books/7")
    assert response.status_code == 404
    service.get_book.assert_called_once_with(7)

tests/test_api.py::test_create_and_read_with_a_fake PASSED
tests/test_api.py::test_duplicate_isbn_is_409 PASSED
tests/test_api.py::test_nested_review_route PASSED
tests/test_api.py::test_web_layer_alone_with_a_mock PASSED
tests/test_service.py::test_add_and_get[fake] PASSED
tests/test_service.py::test_add_and_get[sqlite] PASSED
tests/test_service.py::test_duplicate_isbn_rejected[fake] PASSED
tests/test_service.py::test_duplicate_isbn_rejected[sqlite] PASSED
tests/test_service.py::test_review_for_missing_book[fake] PASSED
tests/test_service.py::test_review_for_missing_book[sqlite] PASSED
tests/test_service.py::test_reviews_are_listed_per_book[fake] PASSED
tests/test_service.py::test_reviews_are_listed_per_book[sqlite] PASSED

backend/
├── pyproject.toml            # has a [tool.fastapi] table
├── alembic.ini
├── app/
│   ├── main.py               # FastAPI(...), CORS, include_router(api_router, prefix=...)
│   ├── models.py             # SQLModel tables and API schemas in one module
│   ├── crud.py               # database functions
│   ├── utils.py
│   ├── api/
│   │   ├── main.py           # api_router that includes each router
│   │   ├── deps.py           # SessionDep, TokenDep, CurrentUser (Annotated aliases)
│   │   └── routes/           # items.py, login.py, users.py, utils.py, private.py
│   ├── core/                 # config.py, db.py, security.py
│   └── alembic/              # migrations
└── tests/                    # api/, crud/, utils/, conftest.py

from fastapi.routing import iter_route_contexts

from app.main import app

for route in app.router.routes:
    print(type(route).__name__, getattr(route, "path", None))

for context in iter_route_contexts(app.router.routes):
    print(sorted(context.methods or []), context.path, context.name)

Route /openapi.json
Route /docs
Route /docs/oauth2-redirect
Route /redoc
_IncludedRouter None
_IncludedRouter None

['GET', 'HEAD'] /openapi.json openapi
['GET', 'HEAD'] /docs swagger_ui_html
['GET', 'HEAD'] /docs/oauth2-redirect swagger_ui_redirect
['GET', 'HEAD'] /redoc redoc_html
['GET'] /api/v1/books/{book_id}/reviews/ list_reviews
['POST'] /api/v1/books/{book_id}/reviews/ create_review
['GET'] /api/v1/books/ list_books
['POST'] /api/v1/books/ create_book
['GET'] /api/v1/books/{book_id} read_book
['GET'] /admin/stats stats