Chapter 9: Application Services — Coordinating Use Cases

So far we have command handlers for processing commands and event handlers for reacting to events. But real applications also need a synchronous coordination layer — something that handles a user's request end-to-end. Application services fill this role.

What Are Application Services?

An application service coordinates a specific use case. Each method typically maps to one API endpoint or user action:

• CatalogService.add_book(...) → POST /books
• CatalogService.get_book(id) → GET /books/:id
• CatalogService.search_books(...) → GET /books?author=...

Application services are a thin orchestration layer. They do not contain business logic — they delegate to aggregates, repositories, and domain services.

Building the CatalogService

    @use_case
    def add_book(
        self,
        title: str,
        author: str,
        isbn: str = None,
        price_amount: float = 0.0,
        description: str = "",
    ) -> Identifier:
        """Add a new book to the catalog."""
        book = Book(
            title=title,
            author=author,
            isbn=isbn,
            price=Money(amount=price_amount),
            description=description,
        )
        book.add_to_catalog()
        current_domain.repository_for(Book).add(book)
        return book.id

    @use_case
    def get_book(self, book_id: Identifier) -> Book:
        """Retrieve a book by its ID."""
        return current_domain.repository_for(Book).get(book_id)

    @use_case
    def search_books(self, **filters) -> list:
        """Search for books matching the given filters."""
        query = current_domain.repository_for(Book)._dao.query
        for field, value in filters.items():
            query = query.filter(**{field: value})
        return query.all().items

Key points:

• @domain.application_service(part_of=Book) registers the service with the Book aggregate.
• @use_case wraps each method in a Unit of Work — automatic transaction management, just like command handlers.
• Methods are plain Python — create aggregates, use repositories, return results.

The @use_case Decorator

The @use_case decorator does two important things:

1. Wraps the method in a Unit of Work — if the method succeeds, changes are committed. If it raises an exception, everything rolls back.
2. Provides a clear boundary — each use case is a self-contained operation.

Without @use_case, you would need to manage transactions manually:

# Without @use_case (manual)
def add_book(self, ...):
    with UnitOfWork() as uow:
        book = Book(...)
        repo.add(book)

# With @use_case (automatic)
@use_case
def add_book(self, ...):
    book = Book(...)
    repo.add(book)

Application Services vs Command Handlers

Both coordinate state changes. When should you use which?

Application Services	Command Handlers
Synchronous, returns result immediately	Can run asynchronously via server
Called directly from API/UI layer	Dispatched via domain.process()
Good for request-response workflows	Good for fire-and-forget commands
Orchestrates multiple operations	Processes a single command

They are not mutually exclusive. An application service might call domain.process() internally, or a command handler might use shared domain logic.

Choosing the Right Approach

For a typical web API:

• Use application services for operations where the client needs an immediate response (e.g., creating a resource and returning its ID).
• Use command handlers for operations that can be queued and processed in the background (e.g., bulk imports, long-running tasks).

Using the Service

if __name__ == "__main__":
    with domain.domain_context():
        catalog = CatalogService()

        # Add books through the application service
        print("=== Adding Books via CatalogService ===")
        id1 = catalog.add_book(
            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"Added: The Great Gatsby (ID: {id1})")

        id2 = catalog.add_book(
            title="Brave New World",
            author="Aldous Huxley",
            isbn="9780060850524",
            price_amount=14.99,
        )
        print(f"Added: Brave New World (ID: {id2})")

        id3 = catalog.add_book(
            title="1984",
            author="George Orwell",
            isbn="9780451524935",
            price_amount=11.99,
        )
        print(f"Added: 1984 (ID: {id3})")

        # Get a specific book
        print("\n=== Retrieving a Book ===")
        book = catalog.get_book(id1)
        print(f"Found: {book.title} by {book.author}, ${book.price.amount}")

        # Search for books by author
        print("\n=== Searching Books ===")
        all_books = catalog.search_books()
        print(f"Total books: {len(all_books)}")
        for b in all_books:
            print(f"  - {b.title} by {b.author}")

        # Verify
        assert len(all_books) == 3
        assert book.title == "The Great Gatsby"

Run it:

$ python bookshelf.py
=== Adding Books via CatalogService ===
Added: The Great Gatsby (ID: a3b2c1d0-...)
Added: Brave New World (ID: e5f6g7h8-...)
Added: 1984 (ID: i9j0k1l2-...)

=== Retrieving a Book ===
Found: The Great Gatsby by F. Scott Fitzgerald, $12.99

=== Searching Books ===
Total books: 3
  - The Great Gatsby by F. Scott Fitzgerald
  - Brave New World by Aldous Huxley
  - 1984 by George Orwell

All checks passed!

Full Source

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

domain = Domain()

domain.config["command_processing"] = "sync"
domain.config["event_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()

    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,
            )
        )


@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.application_service(part_of=Book)
class CatalogService:
    @use_case
    def add_book(
        self,
        title: str,
        author: str,
        isbn: str = None,
        price_amount: float = 0.0,
        description: str = "",
    ) -> Identifier:
        """Add a new book to the catalog."""
        book = Book(
            title=title,
            author=author,
            isbn=isbn,
            price=Money(amount=price_amount),
            description=description,
        )
        book.add_to_catalog()
        current_domain.repository_for(Book).add(book)
        return book.id

    @use_case
    def get_book(self, book_id: Identifier) -> Book:
        """Retrieve a book by its ID."""
        return current_domain.repository_for(Book).get(book_id)

    @use_case
    def search_books(self, **filters) -> list:
        """Search for books matching the given filters."""
        query = current_domain.repository_for(Book)._dao.query
        for field, value in filters.items():
            query = query.filter(**{field: value})
        return query.all().items




domain.init(traverse=False)


if __name__ == "__main__":
    with domain.domain_context():
        catalog = CatalogService()

        # Add books through the application service
        print("=== Adding Books via CatalogService ===")
        id1 = catalog.add_book(
            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"Added: The Great Gatsby (ID: {id1})")

        id2 = catalog.add_book(
            title="Brave New World",
            author="Aldous Huxley",
            isbn="9780060850524",
            price_amount=14.99,
        )
        print(f"Added: Brave New World (ID: {id2})")

        id3 = catalog.add_book(
            title="1984",
            author="George Orwell",
            isbn="9780451524935",
            price_amount=11.99,
        )
        print(f"Added: 1984 (ID: {id3})")

        # Get a specific book
        print("\n=== Retrieving a Book ===")
        book = catalog.get_book(id1)
        print(f"Found: {book.title} by {book.author}, ${book.price.amount}")

        # Search for books by author
        print("\n=== Searching Books ===")
        all_books = catalog.search_books()
        print(f"Total books: {len(all_books)}")
        for b in all_books:
            print(f"  - {b.title} by {b.author}")

        # Verify
        assert len(all_books) == 3
        assert book.title == "The Great Gatsby"
        print("\nAll checks passed!")

Summary

In this chapter you learned:

• Application services coordinate use cases, acting as the bridge between the API layer and the domain.
• @use_case wraps methods in a Unit of Work for automatic transaction management.
• Application services are a thin orchestration layer — they delegate to aggregates and repositories.
• Use application services for synchronous request-response flows, command handlers for asynchronous processing.

Sometimes business logic spans multiple aggregates — "place an order" involves both the Order and Inventory. In the next chapter we will build a domain service to handle this cross-aggregate coordination.

Next

Chapter 10: Domain Services →