Chapter 9: Structuring the Project

Our bookshelf.py file has grown to hundreds of lines — aggregates, value objects, commands, handlers, events, projections, and projectors all in one file. Before we add an API layer and more features, we need a proper project structure.

Why Restructure Now?

A single file works for learning, but a real application needs separation:

• Commands and handlers grow independently of the domain model.
• Projections and projectors change at a different pace than aggregates.
• An API layer needs clean imports from organized modules.
• Tests mirror the source structure.

The Target Layout

bookshelf/
  __init__.py          # Domain instance
  models.py            # Aggregates, entities, value objects
  commands.py          # Commands
  events.py            # Domain events
  handlers.py          # Command handlers and event handlers
  projections.py       # Projections and projectors
domain.toml            # Configuration
tests/
  conftest.py
  test_commands.py
  test_invariants.py

Creating the Package

The Domain Instance

The domain lives in bookshelf/__init__.py:

# bookshelf/__init__.py
from protean import Domain

domain = Domain("bookshelf")

Models

Aggregates, entities, and value objects go in bookshelf/models.py:

# bookshelf/models.py
from enum import Enum

from bookshelf import domain

from protean.fields import (
    Float,
    HasMany,
    Identifier,
    Integer,
    String,
    Text,
    ValueObject,
)


@domain.value_object
class Money:
    currency: String(max_length=3, default="USD")
    amount: Float(required=True)


@domain.value_object
class Address:
    street: String(max_length=200)
    city: String(max_length=100)
    state: String(max_length=50)
    zip_code: String(max_length=10)
    country: String(max_length=50, default="US")


class OrderStatus(Enum):
    PENDING = "PENDING"
    CONFIRMED = "CONFIRMED"
    SHIPPED = "SHIPPED"


@domain.aggregate
class Book:
    title: String(max_length=200, required=True)
    author: String(max_length=150, required=True)
    isbn: String(max_length=13)
    price = ValueObject(Money)
    description: Text()

    def add_to_catalog(self):
        self.raise_(
            BookAdded(
                book_id=self.id,
                title=self.title,
                author=self.author,
                price_amount=self.price.amount if self.price else 0,
            )
        )

    def update_price(self, new_price: float):
        self.price = Money(amount=new_price)
        self.raise_(BookPriceUpdated(book_id=self.id, new_price=new_price))


@domain.aggregate
class Order:
    customer_name: String(max_length=150, required=True)
    status: String(
        max_length=20, choices=OrderStatus, default=OrderStatus.PENDING.value
    )
    items = HasMany("OrderItem")

    def confirm(self):
        self.status = OrderStatus.CONFIRMED.value
        self.raise_(OrderConfirmed(order_id=self.id, customer_name=self.customer_name))

    def ship(self):
        self.status = OrderStatus.SHIPPED.value
        self.raise_(OrderShipped(order_id=self.id, customer_name=self.customer_name))


@domain.entity(part_of=Order)
class OrderItem:
    book_title: String(max_length=200, required=True)
    quantity: Integer(required=True)
    unit_price = ValueObject(Money)


@domain.aggregate
class Inventory:
    book_id: Identifier(required=True)
    title: String(max_length=200, required=True)
    quantity: Integer(default=0)

    def adjust_stock(self, amount: int):
        self.quantity += amount

Events

Domain events go in bookshelf/events.py:

# bookshelf/events.py
from bookshelf import domain
from bookshelf.models import Book, Order

from protean.fields import Float, Identifier, String


@domain.event(part_of=Book)
class BookAdded:
    book_id: Identifier(required=True)
    title: String(max_length=200, required=True)
    author: String(max_length=150, required=True)
    price_amount: Float()


@domain.event(part_of=Book)
class BookPriceUpdated:
    book_id: Identifier(required=True)
    new_price: Float(required=True)


@domain.event(part_of=Order)
class OrderConfirmed:
    order_id: Identifier(required=True)
    customer_name: String(max_length=150, required=True)


@domain.event(part_of=Order)
class OrderShipped:
    order_id: Identifier(required=True)
    customer_name: String(max_length=150, required=True)

Commands

Commands go in bookshelf/commands.py:

# bookshelf/commands.py
from bookshelf import domain
from bookshelf.models import Book, Order

from protean.fields import Float, Identifier, Integer, String, Text


@domain.command(part_of=Book)
class AddBook:
    title: String(max_length=200, required=True)
    author: String(max_length=150, required=True)
    isbn: String(max_length=13)
    price_amount: Float(required=True)
    description: Text()


@domain.command(part_of=Order)
class PlaceOrder:
    customer_name: String(max_length=150, required=True)
    book_title: String(max_length=200, required=True)
    quantity: Integer(required=True)
    unit_price_amount: Float(required=True)


@domain.command(part_of=Order)
class ConfirmOrder:
    order_id: Identifier(required=True)


@domain.command(part_of=Order)
class ShipOrder:
    order_id: Identifier(required=True)

Handlers

Command handlers and event handlers go in bookshelf/handlers.py:

# bookshelf/handlers.py
from bookshelf import domain
from bookshelf.commands import AddBook, ConfirmOrder, PlaceOrder, ShipOrder
from bookshelf.events import BookAdded, OrderConfirmed, OrderShipped
from bookshelf.models import Book, Inventory, Money, Order, OrderItem

from protean import handle
from protean.fields import Identifier
from protean.utils.globals import current_domain


@domain.command_handler(part_of=Book)
class BookCommandHandler:
    @handle(AddBook)
    def add_book(self, command: AddBook) -> Identifier:
        book = Book(
            title=command.title,
            author=command.author,
            isbn=command.isbn,
            price=Money(amount=command.price_amount),
            description=command.description,
        )
        book.add_to_catalog()
        current_domain.repository_for(Book).add(book)
        return book.id


@domain.command_handler(part_of=Order)
class OrderCommandHandler:
    @handle(PlaceOrder)
    def place_order(self, command: PlaceOrder) -> Identifier:
        order = Order(
            customer_name=command.customer_name,
            items=[
                OrderItem(
                    book_title=command.book_title,
                    quantity=command.quantity,
                    unit_price=Money(amount=command.unit_price_amount),
                ),
            ],
        )
        current_domain.repository_for(Order).add(order)
        return order.id

    @handle(ConfirmOrder)
    def confirm_order(self, command: ConfirmOrder) -> None:
        repo = current_domain.repository_for(Order)
        order = repo.get(command.order_id)
        order.confirm()
        repo.add(order)

    @handle(ShipOrder)
    def ship_order(self, command: ShipOrder) -> None:
        repo = current_domain.repository_for(Order)
        order = repo.get(command.order_id)
        order.ship()
        repo.add(order)


@domain.event_handler(part_of=Book)
class BookEventHandler:
    @handle(BookAdded)
    def on_book_added(self, event: BookAdded):
        inventory = Inventory(
            book_id=event.book_id,
            title=event.title,
            quantity=10,
        )
        current_domain.repository_for(Inventory).add(inventory)


@domain.event_handler(part_of=Order)
class OrderEventHandler:
    @handle(OrderConfirmed)
    def on_order_confirmed(self, event: OrderConfirmed):
        print(
            f"  [Notification] Order {event.order_id} confirmed for {event.customer_name}"
        )

    @handle(OrderShipped)
    def on_order_shipped(self, event: OrderShipped):
        print(
            f"  [Notification] Order {event.order_id} shipped to {event.customer_name}"
        )

Projections

Projections and projectors go in bookshelf/projections.py:

# bookshelf/projections.py
from bookshelf import domain
from bookshelf.events import BookAdded, BookPriceUpdated
from bookshelf.models import Book

from protean.core.projector import on
from protean.fields import Float, Identifier, String


@domain.projection
class BookCatalog:
    book_id: Identifier(identifier=True, required=True)
    title: String(max_length=200, required=True)
    author: String(max_length=150, required=True)
    price: Float()
    isbn: String(max_length=13)


@domain.projector(projector_for=BookCatalog, aggregates=[Book])
class BookCatalogProjector:
    @on(BookAdded)
    def on_book_added(self, event: BookAdded):
        catalog_entry = BookCatalog(
            book_id=event.book_id,
            title=event.title,
            author=event.author,
            price=event.price_amount,
            isbn=getattr(event, "isbn", ""),
        )
        current_domain.repository_for(BookCatalog).add(catalog_entry)

    @on(BookPriceUpdated)
    def on_price_updated(self, event: BookPriceUpdated):
        repo = current_domain.repository_for(BookCatalog)
        entry = repo.get(event.book_id)
        entry.price = event.new_price
        repo.add(entry)

Domain Auto-Discovery

Notice that we no longer pass traverse=False to domain.init(). With a proper package structure, Protean auto-discovers all domain elements by scanning the package:

# In __init__.py
domain.init()  # traverse=True by default — scans bookshelf/ for elements

This finds all @domain.aggregate, @domain.command, @domain.event, etc. decorators across every module in the bookshelf/ package.

The Configuration File

Move domain.toml to the project root (next to the bookshelf/ package):

debug = true
event_processing = "sync"
command_processing = "sync"

[databases.default]
provider = "postgresql"
database_uri = "${DATABASE_URL|postgresql://postgres:postgres@localhost:5432/bookshelf}"

Using CLI Tools

With a package structure, Protean CLI tools need to know where the domain lives. Use the --domain flag:

$ protean shell --domain bookshelf
>>> from bookshelf.models import Book
>>> domain.repository_for(Book).query.all().total
3

Or set the PROTEAN_DOMAIN environment variable so you don't have to pass --domain every time:

$ export PROTEAN_DOMAIN=bookshelf
$ protean shell
$ protean database setup

Verifying the Structure

Run the application to make sure everything still works:

$ python -c "from bookshelf import domain; domain.init(); print('Domain initialized with', len(domain.registry._elements), 'elements')"

All domain elements should be discovered and registered just as before.

What We Built

• A proper Python package with separate modules for models, commands, events, handlers, and projections.
• Domain auto-discovery — domain.init() scans the package automatically.
• CLI integration with --domain and PROTEAN_DOMAIN.
• The same functionality as before, but organized for growth.

In the next chapter, we will add a FastAPI web layer to expose our domain through HTTP endpoints.

Next

Chapter 10: Exposing the Domain Through an API →