Implementing DDD Building Blocks

Hello! Welcome back to our course on distributed systems architecture.

In our last lesson, we explored the eventsourcing library's mechanisms for event propagation. We saw how the NotificationLog provides a reliable, ordered stream of all state changes and how this can be used for in-process communication with the System class or to publish events to an external message broker.

Today, we will implement the source of those events: the core domain model. This lesson directly addresses the learning outcome:

Implement aggregates, commands, and events for a sample application using the framework's abstractions.

We will shift our focus from how events are consumed to how they are created. You will learn to define aggregates, which are the heart of a domain model in Domain-Driven Design (DDD), and use the library's powerful abstractions to have them produce events in response to commands. We will build a simple but practical BankingApplication to see these concepts in action.

1. Aggregates and Events in the eventsourcing Library

From your background in software and systems, you'll know that managing state and business rules is a central challenge. In event sourcing and DDD, the Aggregate is the primary pattern for this. An aggregate is a cluster of domain objects that can be treated as a single unit. It acts as a transactional consistency boundary, meaning it is responsible for enforcing its own business rules (invariants) before any state change is accepted.

When a command is executed on an aggregate, it doesn't just change its state directly. Instead, it validates the command against its current state and, if valid, produces one or more events. These immutable events are then stored and used to reconstruct the aggregate's state.

The eventsourcing library provides two key abstractions for this:

• The Aggregate base class.
• The @event decorator.

Let's see how they work together.

2. Practical Implementation: The Account Aggregate

Now, let's apply those concepts to a domain more aligned with your experience in finance: a simple bank account. We'll create an Account aggregate that can be opened, have funds deposited, and have funds withdrawn.

Here is the implementation of the Account aggregate. Note the use of Decimal for financial values to avoid floating-point inaccuracies.

import uuid
from decimal import Decimal
from eventsourcing.domain import Aggregate, event

class Account(Aggregate):
    """A bank account aggregate."""

    def __init__(self, owner: str, initial_balance: Decimal = Decimal('0')):
        self.owner = owner
        self.balance = initial_balance
        self.is_closed = False

    @event('Opened')
    @classmethod
    def open(cls, owner: str, initial_balance: Decimal):
        """Class method to create a new account."""
        return cls(owner=owner, initial_balance=initial_balance)

    @event('FundsDeposited')
    def deposit(self, amount: Decimal):
        """Deposit funds into the account."""
        if self.is_closed:
            raise ValueError("Account is closed.")
        if amount <= 0:
            raise ValueError("Deposit amount must be positive.")
        self.balance += amount

    @event('FundsWithdrawn')
    def withdraw(self, amount: Decimal):
        """Withdraw funds from the account."""
        if self.is_closed:
            raise ValueError("Account is closed.")
        if amount <= 0:
            raise ValueError("Withdrawal amount must be positive.")
        if self.balance < amount:
            raise InsufficientFundsError("Insufficient funds.")
        self.balance -= amount

class InsufficientFundsError(Exception):
    pass

Let's break this down:

• Account(Aggregate): Our class inherits from the library's Aggregate.
• @event('Opened'): We decorate a class method open to serve as our factory. When Account.open(...) is called, the library creates an Account.Opened event. The event's mutate() method then calls the __init__ constructor to create the aggregate instance. This is the two-stage construction process from the tutorial.
• @event('FundsDeposited'): The deposit method is decorated. When called on an Account instance, it first runs the method body. If it completes without an exception, a FundsDeposited event is generated and queued. The method parameters (e.g., amount) are automatically captured as attributes on the event object.
• Business Logic: The withdraw method contains an invariant check (self.balance < amount). If this check fails, it raises an InsufficientFundsError. Crucially, no event is generated and the state remains unchanged. This is how aggregates protect the integrity of your domain.

3. The Application Layer: Handling Commands

Aggregates contain the business logic, but something needs to orchestrate the process of loading them, executing commands, and saving the resulting events. This is the role of the Application layer. In the eventsourcing library, this is handled by subclassing the Application class.

This pattern directly implements the "Command Handler" concept from CQRS and DDD.

Now, let's see how to implement this service layer using the library.

4. Practical Implementation: The BankingApplication

Following the pattern from the documentation, we can create a BankingApplication to expose the capabilities of our Account aggregate. The methods on this class are our Commands.

from eventsourcing.application import Application

class BankingApplication(Application):
    def open_account(self, owner: str, initial_balance: Decimal) -> uuid.UUID:
        """Command to open a new account."""
        account = Account.open(owner=owner, initial_balance=initial_balance)
        self.save(account)
        return account.id

    def deposit_funds(self, account_id: uuid.UUID, amount: Decimal):
        """Command to deposit funds."""
        account: Account = self.repository.get(account_id)
        account.deposit(amount)
        self.save(account)

    def withdraw_funds(self, account_id: uuid.UUID, amount: Decimal):
        """Command to withdraw funds."""
        account: Account = self.repository.get(account_id)
        account.withdraw(amount)
        self.save(account)

    def get_balance(self, account_id: uuid.UUID) -> Decimal:
        """Query to get the current balance."""
        account: Account = self.repository.get(account_id)
        return account.balance

Notice the clear and consistent pattern:

1. open_account (Creation): Calls the aggregate's factory method (Account.open), then calls self.save() to persist the initial Opened event.
2. deposit_funds (Update):
   • Load: self.repository.get(account_id) retrieves the event stream for the given ID and replays it to reconstruct the Account object in its current state.
   • Execute: account.deposit(amount) calls the aggregate method, which validates the command and generates a new event.
   • Save: self.save(account) collects the pending event (FundsDeposited) and appends it to the event store.
3. get_balance (Query): This is a read-only operation. It simply loads the aggregate to its current state and returns a value. No events are generated, and self.save() is not called.

5. Exercise: Extending the Domain

To solidify your understanding, let's add a new feature: the ability to close an account.

Your task:

1. Add an is_closed boolean attribute to the Account aggregate's __init__ method, defaulting to False.
2. Modify deposit() and withdraw() to raise an exception if the account is closed.
3. Create a new command method on the Account aggregate called close().
   • It should be decorated to produce an AccountClosed event.
   • It must enforce the business rule: an account can only be closed if its balance is exactly zero.
   • When the AccountClosed event is applied, it should set is_closed to True.
4. Add a corresponding close_account(account_id: uuid.UUID) command method to the BankingApplication.

# In the Account aggregate class:

# ... (inside __init__)
# self.is_closed = False # Already added this earlier

@event('AccountClosed')
def close(self):
    """Close the account."""
    if self.is_closed:
        raise ValueError("Account is already closed.")
    if self.balance != Decimal('0'):
        raise ValueError("Cannot close account with a non-zero balance.")
    self.is_closed = True

# In the BankingApplication class:

def close_account(self, account_id: uuid.UUID):
    """Command to close an account."""
    account: Account = self.repository.get(account_id)
    account.close()
    self.save(account)

# Example Usage:
# app = BankingApplication()
# account_id = app.open_account("Alice", Decimal('100'))
# app.withdraw_funds(account_id, Decimal('100'))
# app.close_account(account_id)
#
# try:
#     app.deposit_funds(account_id, Decimal('50'))
# except ValueError as e:
#     print(f"Caught expected error: {e}") # "Account is closed."

Conclusion

In this lesson, you have implemented the fundamental components of an event-sourced application using the eventsourcing library. You have seen how to model a domain entity as an aggregate, enforce business rules, and generate events that capture every state change.

Key Takeaways:

• Aggregates are classes that inherit from eventsourcing.domain.Aggregate and encapsulate state and business logic.
• The @event decorator is the primary mechanism for turning a method call into a persisted event. It separates the intention (calling the method) from the state change (the body of the method, applied via the event).
• Business rules and invariants are enforced within aggregate methods. If a rule is violated, an exception is raised, and no event is created.
• The Application class serves as the command handler layer. It orchestrates the load -> execute -> save cycle for interacting with aggregates.

Preview of the Next Lesson:

Our BankingApplication currently runs entirely in memory. If you restart the application, all the accounts and their history are lost. The next lesson, "Configure the framework to use a separate persistence service (e.g., PostgreSQL) for event storage," will address this. You will see how, with a simple configuration change, we can make our application durable by storing its events in a real database, without changing any of our domain or application code.