Chapter 5: Commands and Handlers

In this chapter we will add an AddBook command and a handler that processes it, so books are added through a formal command interface instead of direct aggregate creation.

Defining a Command

So far we have created aggregates directly. In a real application, state changes arrive as commands — formal requests to do something, named with imperative verbs:

@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)
    price_currency: String(max_length=3, default="USD")
    description: Text()

A command is an immutable data object — it carries the intent ("add this book") and the data needed to fulfill it.

The Command Handler

A command handler receives the command and orchestrates the state change:

@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,
                currency=command.price_currency,
            ),
            description=command.description,
        )
        current_domain.repository_for(Book).add(book)
        return book.id

Notice the pattern: receive command, create aggregate, persist it, return the result. Each handler method runs in a transaction automatically.

Dispatching Commands

To dispatch a command, use domain.process():

if __name__ == "__main__":
    with domain.domain_context():
        # Process a command to add a book
        book_id = domain.process(
            AddBook(
                title="The Great Gatsby",
                author="F. Scott Fitzgerald",
                isbn="9780743273565",
                price_amount=12.99,
                description="A story of the mysteriously wealthy Jay Gatsby.",
            )
        )

We set command_processing = "sync" so commands are processed immediately and domain.process() returns the handler's result.

Run it:

$ python bookshelf.py
Book added with ID: a3b2c1d0-...
Retrieved: The Great Gatsby by F. Scott Fitzgerald
Price: $12.99 USD

Total books: 2
  - The Great Gatsby
  - Brave New World

All checks passed!

Notice that we never touched the repository directly — the command handler did that for us. This separation means the same command can later be processed asynchronously by a background server.

What We Built

An AddBook command — an immutable intent object.
A BookCommandHandler — receives the command, creates and persists the aggregate.
domain.process() — dispatches commands to their handlers.

In the next chapter, we will add domain events and event handlers so the system can react automatically when things happen.

Full Source

from protean import Domain, handle
from protean.fields import Float, Identifier, String, Text, ValueObject
from protean.utils.globals import current_domain

domain = Domain()

domain.config["command_processing"] = "sync"


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


@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()


@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)
    price_currency: String(max_length=3, default="USD")
    description: Text()




@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,
                currency=command.price_currency,
            ),
            description=command.description,
        )
        current_domain.repository_for(Book).add(book)
        return book.id




domain.init(traverse=False)


if __name__ == "__main__":
    with domain.domain_context():
        # Process a command to add a book
        book_id = domain.process(
            AddBook(
                title="The Great Gatsby",
                author="F. Scott Fitzgerald",
                isbn="9780743273565",
                price_amount=12.99,
                description="A story of the mysteriously wealthy Jay Gatsby.",
            )
        )
        print(f"Book added with ID: {book_id}")

        # The book is now in the repository
        book = current_domain.repository_for(Book).get(book_id)
        print(f"Retrieved: {book.title} by {book.author}")
        print(f"Price: ${book.price.amount} {book.price.currency}")

        # Add another book
        book_id_2 = domain.process(
            AddBook(
                title="Brave New World",
                author="Aldous Huxley",
                isbn="9780060850524",
                price_amount=14.99,
                description="A dystopian novel set in a futuristic World State.",
            )
        )

        # Verify both books exist
        all_books = current_domain.repository_for(Book)._dao.query.all()
        print(f"\nTotal books: {all_books.total}")
        for b in all_books.items:
            print(f"  - {b.title}")

        # Verify
        assert all_books.total == 2
        print("\nAll checks passed!")