Chapter 10: Exposing the Domain Through an API

Our domain logic works and the project is organized — but nobody can use the bookstore yet. In this chapter we will build a FastAPI web layer that translates HTTP requests into domain commands and projection queries.

Thin Endpoints

API endpoints are thin adapters at the boundary of the domain. They translate HTTP requests into commands, hand them to domain.process(), and translate the results back to HTTP responses. All business logic stays in the domain layer.

Setting Up FastAPI

Install FastAPI and Uvicorn:

pip install fastapi uvicorn

Create bookshelf/api.py:

from fastapi import FastAPI
from pydantic import BaseModel

from protean.integrations.fastapi import (
    DomainContextMiddleware,
    register_exception_handlers,
)

from bookshelf import domain
from bookshelf.commands import AddBook, ConfirmOrder, PlaceOrder, ShipOrder
from bookshelf.projections import BookCatalog

domain.init()

app = FastAPI(title="Bookshelf API")

app.add_middleware(
    DomainContextMiddleware,
    route_domain_map={"/": domain},
)
register_exception_handlers(app)

DomainContextMiddleware ensures every request runs inside the correct domain context. register_exception_handlers() maps domain exceptions (like ObjectNotFoundError, ValidationError) to standard HTTP error responses (404, 400, etc.).

Write Endpoints — Commands

Each write endpoint translates an HTTP request into a command and dispatches it:

class AddBookRequest(BaseModel):
    title: str
    author: str
    isbn: str | None = None
    price_amount: float
    description: str | None = None


class PlaceOrderRequest(BaseModel):
    customer_name: str
    book_title: str
    quantity: int
    unit_price_amount: float


@app.post("/books")
def add_book(request: AddBookRequest):
    book_id = domain.process(
        AddBook(
            title=request.title,
            author=request.author,
            isbn=request.isbn,
            price_amount=request.price_amount,
            description=request.description,
        )
    )
    return {"book_id": str(book_id)}


@app.post("/orders")
def place_order(request: PlaceOrderRequest):
    order_id = domain.process(
        PlaceOrder(
            customer_name=request.customer_name,
            book_title=request.book_title,
            quantity=request.quantity,
            unit_price_amount=request.unit_price_amount,
        )
    )
    return {"order_id": str(order_id)}


@app.post("/orders/{order_id}/confirm")
def confirm_order(order_id: str):
    domain.process(ConfirmOrder(order_id=order_id))
    return {"status": "confirmed"}


@app.post("/orders/{order_id}/ship")
def ship_order(order_id: str):
    domain.process(ShipOrder(order_id=order_id))
    return {"status": "shipped"}

The pattern is always the same:

Accept a Pydantic request model.
Build a domain command from the request data.
Call domain.process(command).
Return the result.

Read Endpoints — Querying Projections

Read endpoints query projections using domain.query_for():

@app.get("/catalog")
def browse_catalog():
    results = domain.query_for(BookCatalog).all()
    return {
        "entries": [
            {
                "book_id": str(entry.book_id),
                "title": entry.title,
                "author": entry.author,
                "price": entry.price,
                "isbn": entry.isbn,
            }
            for entry in results.items
        ],
        "total": results.total,
    }


@app.get("/catalog/{book_id}")
def get_catalog_entry(book_id: str):
    entry = domain.repository_for(BookCatalog).get(book_id)
    return {
        "book_id": str(entry.book_id),
        "title": entry.title,
        "author": entry.author,
        "price": entry.price,
        "isbn": entry.isbn,
    }

domain.query_for(BookCatalog) returns a ReadOnlyQuerySet that supports filtering, sorting, and pagination — but blocks any writes.

Running the API

Start the server:

$ uvicorn bookshelf.api:app --reload
INFO:     Uvicorn running on http://127.0.0.1:8000

Test it with curl:

# Add a book
$ curl -X POST http://localhost:8000/books \
  -H "Content-Type: application/json" \
  -d '{"title": "The Great Gatsby", "author": "F. Scott Fitzgerald", "price_amount": 12.99}'

{"book_id": "a3b2c1d0-..."}

# Browse the catalog
$ curl http://localhost:8000/catalog

{"entries": [{"book_id": "a3b2c1d0-...", "title": "The Great Gatsby", ...}], "total": 1}

# Place an order
$ curl -X POST http://localhost:8000/orders \
  -H "Content-Type: application/json" \
  -d '{"customer_name": "Alice", "book_title": "The Great Gatsby", "quantity": 1, "unit_price_amount": 12.99}'

{"order_id": "e5f6g7h8-..."}

Visit http://localhost:8000/docs for the interactive Swagger documentation that FastAPI generates automatically.

What We Built

A FastAPI application with DomainContextMiddleware and automatic error handling.
Write endpoints that translate HTTP requests into domain commands.
Read endpoints that query projections with domain.query_for().
A running web server that exposes the bookstore over HTTP.

In the next chapter, we will add tests for everything we have built so far — domain logic, command flows, and API endpoints.

Full Source

from fastapi import FastAPI
from pydantic import BaseModel

from protean.integrations.fastapi import (
    DomainContextMiddleware,
    register_exception_handlers,
)

from bookshelf import domain
from bookshelf.commands import AddBook, ConfirmOrder, PlaceOrder, ShipOrder
from bookshelf.projections import BookCatalog

domain.init()

app = FastAPI(title="Bookshelf API")

app.add_middleware(
    DomainContextMiddleware,
    route_domain_map={"/": domain},
)
register_exception_handlers(app)


class AddBookRequest(BaseModel):
    title: str
    author: str
    isbn: str | None = None
    price_amount: float
    description: str | None = None


class PlaceOrderRequest(BaseModel):
    customer_name: str
    book_title: str
    quantity: int
    unit_price_amount: float


@app.post("/books")
def add_book(request: AddBookRequest):
    book_id = domain.process(
        AddBook(
            title=request.title,
            author=request.author,
            isbn=request.isbn,
            price_amount=request.price_amount,
            description=request.description,
        )
    )
    return {"book_id": str(book_id)}


@app.post("/orders")
def place_order(request: PlaceOrderRequest):
    order_id = domain.process(
        PlaceOrder(
            customer_name=request.customer_name,
            book_title=request.book_title,
            quantity=request.quantity,
            unit_price_amount=request.unit_price_amount,
        )
    )
    return {"order_id": str(order_id)}


@app.post("/orders/{order_id}/confirm")
def confirm_order(order_id: str):
    domain.process(ConfirmOrder(order_id=order_id))
    return {"status": "confirmed"}


@app.post("/orders/{order_id}/ship")
def ship_order(order_id: str):
    domain.process(ShipOrder(order_id=order_id))
    return {"status": "shipped"}




@app.get("/catalog")
def browse_catalog():
    results = domain.query_for(BookCatalog).all()
    return {
        "entries": [
            {
                "book_id": str(entry.book_id),
                "title": entry.title,
                "author": entry.author,
                "price": entry.price,
                "isbn": entry.isbn,
            }
            for entry in results.items
        ],
        "total": results.total,
    }


@app.get("/catalog/{book_id}")
def get_catalog_entry(book_id: str):
    entry = domain.repository_for(BookCatalog).get(book_id)
    return {
        "book_id": str(entry.book_id),
        "title": entry.title,
        "author": entry.author,
        "price": entry.price,
        "isbn": entry.isbn,
    }