Chapter 4: Business Rules That Never Break

A tester discovers a problem: withdraw $200 from an account with $100 and the withdrawal succeeds — because the validation in withdraw() checks the balance before the @apply handler runs. What if the validation and the state mutation disagree?

In this chapter we will add invariants — rules that are checked after every state mutation, guaranteeing the aggregate is always valid regardless of how it got there.

The Problem with Method-Level Validation

In Chapter 2, our withdraw() method checked if amount > self.balance. This works for the simple case, but it has a subtle issue: the check happens before raise_() calls the @apply handler. If any future code path changes the balance between the check and the event, the validation could be bypassed.

Invariants solve this by validating the aggregate's state after the @apply handler has already mutated it.

Post-Invariants

Add two invariants to the Account aggregate:

    @invariant.post
    def balance_must_not_be_negative(self):
        if self.balance is not None and self.balance < 0:
            raise ValidationError(
                {"balance": ["Insufficient funds: balance cannot be negative"]}
            )

    @invariant.post
    def closed_account_must_have_zero_balance(self):
        if self.status == "CLOSED" and self.balance != 0:
            raise ValidationError(
                {"status": ["Cannot close account with non-zero balance"]}
            )

Here is how they work:

1. When raise_() is called on an event-sourced aggregate, Protean:

   • Records the event
   • Calls the matching @apply handler (state is mutated)
   • Runs all @invariant.post methods
   • If any invariant raises ValidationError, the event is rejected and state is rolled back

2. balance_must_not_be_negative ensures no operation can leave the account with a negative balance. We no longer need the manual check in withdraw() — the invariant handles it.

3. closed_account_must_have_zero_balance ensures an account cannot be closed while funds remain. This rule would be difficult to enforce with method-level checks alone because it spans two fields (status and balance).

Closing an Account

We also add an AccountClosed event and a close() method:

@domain.event(part_of="Account")
class AccountClosed:
    account_id: Identifier(required=True)
    reason: String()

    def close(self, reason: str = None) -> None:
        self.raise_(
            AccountClosed(
                account_id=str(self.id),
                reason=reason,
            )
        )

And the corresponding @apply handler:

    @apply
    def on_account_closed(self, event: AccountClosed):
        self.status = "CLOSED"

Invariants in Action

Let's exercise both invariants:

if __name__ == "__main__":
    with domain.domain_context():
        # Open an account with $100
        account_id = domain.process(
            OpenAccount(
                account_number="ACC-001",
                holder_name="Alice Johnson",
                opening_deposit=100.00,
            )
        )
        print(f"Account opened: {account_id}")

        # Try to withdraw $200 — the invariant will catch this
        try:
            domain.process(MakeWithdrawal(account_id=account_id, amount=200.00))
        except ValidationError as e:
            print(f"Overdraft rejected: {e.messages}")

        # Try to close with a non-zero balance — invariant catches this too
        try:
            domain.process(
                CloseAccount(account_id=account_id, reason="Customer request")
            )
        except ValidationError as e:
            print(f"Close rejected: {e.messages}")

        # Withdraw all funds first, then close
        domain.process(MakeWithdrawal(account_id=account_id, amount=100.00))
        domain.process(CloseAccount(account_id=account_id, reason="Customer request"))

        repo = domain.repository_for(Account)
        account = repo.get(account_id)
        print(f"\nAccount status: {account.status}")
        print(f"Balance: ${account.balance:.2f}")

        assert account.status == "CLOSED"
        assert account.balance == 0.0
        print("\nAll checks passed!")

Run it:

$ python fidelis.py
Account opened: acc-001-uuid...
Overdraft rejected: {'balance': ['Insufficient funds: balance cannot be negative']}
Close rejected: {'status': ['Cannot close account with non-zero balance']}

Account status: CLOSED
Balance: $0.00

All checks passed!

The overdraft attempt was caught by balance_must_not_be_negative — the @apply handler reduced the balance to -$100, the invariant detected the violation, and Protean rolled back the event.

The close attempt was caught by closed_account_must_have_zero_balance — the @apply handler set status = "CLOSED", but the invariant saw the balance was still $100 and rejected the transition.

Why Invariants Are Better

Invariants have important advantages over method-level checks:

Method-Level Checks	Invariants
Run before the event	Run after the @apply handler
Check preconditions	Validate postconditions
Can be bypassed by new code paths	Always enforced, every time
Do not run during replay	Also enforced during replay
Must be duplicated across methods	Defined once, applied everywhere

Always Valid

Post-invariants guarantee that the aggregate is in a valid state after every single event. This is the "always valid" principle — there is no window where the aggregate exists in an invalid state.

What We Built

• Post-invariants with @invariant.post that validate state after every @apply handler.
• balance_must_not_be_negative — prevents overdrafts at the aggregate level.
• closed_account_must_have_zero_balance — enforces a multi-field business rule.
• An AccountClosed event and close() method.
• Confidence that business rules are always enforced, regardless of how the aggregate is modified.

Next, we will write proper automated tests using Protean's fluent testing DSL.

Full Source

from protean import Domain, apply, handle, invariant
from protean.exceptions import ValidationError
from protean.fields import Float, Identifier, String
from protean.utils.globals import current_domain

domain = Domain("fidelis")


@domain.event(part_of="Account")
class AccountOpened:
    account_id: Identifier(required=True)
    account_number: String(required=True)
    holder_name: String(required=True)
    opening_deposit: Float(required=True)


@domain.event(part_of="Account")
class DepositMade:
    account_id: Identifier(required=True)
    amount: Float(required=True)
    reference: String()


@domain.event(part_of="Account")
class WithdrawalMade:
    account_id: Identifier(required=True)
    amount: Float(required=True)
    reference: String()


@domain.event(part_of="Account")
class AccountClosed:
    account_id: Identifier(required=True)
    reason: String()




@domain.aggregate(is_event_sourced=True)
class Account:
    account_number: String(max_length=20, required=True)
    holder_name: String(max_length=100, required=True)
    balance: Float(default=0.0)
    status: String(max_length=20, default="ACTIVE")

    @invariant.post
    def balance_must_not_be_negative(self):
        if self.balance is not None and self.balance < 0:
            raise ValidationError(
                {"balance": ["Insufficient funds: balance cannot be negative"]}
            )

    @invariant.post
    def closed_account_must_have_zero_balance(self):
        if self.status == "CLOSED" and self.balance != 0:
            raise ValidationError(
                {"status": ["Cannot close account with non-zero balance"]}
            )

    @classmethod
    def open(cls, account_number: str, holder_name: str, opening_deposit: float):
        account = cls._create_new()
        account.raise_(
            AccountOpened(
                account_id=str(account.id),
                account_number=account_number,
                holder_name=holder_name,
                opening_deposit=opening_deposit,
            )
        )
        return account

    def deposit(self, amount: float, reference: str = None) -> None:
        if amount <= 0:
            raise ValidationError({"amount": ["Deposit amount must be positive"]})
        self.raise_(
            DepositMade(
                account_id=str(self.id),
                amount=amount,
                reference=reference,
            )
        )

    def withdraw(self, amount: float, reference: str = None) -> None:
        if amount <= 0:
            raise ValidationError({"amount": ["Withdrawal amount must be positive"]})
        self.raise_(
            WithdrawalMade(
                account_id=str(self.id),
                amount=amount,
                reference=reference,
            )
        )

    def close(self, reason: str = None) -> None:
        self.raise_(
            AccountClosed(
                account_id=str(self.id),
                reason=reason,
            )
        )

    @apply
    def on_account_opened(self, event: AccountOpened):
        self.id = event.account_id
        self.account_number = event.account_number
        self.holder_name = event.holder_name
        self.balance = event.opening_deposit
        self.status = "ACTIVE"

    @apply
    def on_deposit_made(self, event: DepositMade):
        self.balance += event.amount

    @apply
    def on_withdrawal_made(self, event: WithdrawalMade):
        self.balance -= event.amount

    @apply
    def on_account_closed(self, event: AccountClosed):
        self.status = "CLOSED"



@domain.command(part_of=Account)
class OpenAccount:
    account_number: String(required=True)
    holder_name: String(required=True)
    opening_deposit: Float(required=True)


@domain.command(part_of=Account)
class MakeDeposit:
    account_id: Identifier(required=True)
    amount: Float(required=True)
    reference: String()


@domain.command(part_of=Account)
class MakeWithdrawal:
    account_id: Identifier(required=True)
    amount: Float(required=True)
    reference: String()


@domain.command(part_of=Account)
class CloseAccount:
    account_id: Identifier(required=True)
    reason: String()


@domain.command_handler(part_of=Account)
class AccountCommandHandler:
    @handle(OpenAccount)
    def handle_open_account(self, command: OpenAccount):
        account = Account.open(
            account_number=command.account_number,
            holder_name=command.holder_name,
            opening_deposit=command.opening_deposit,
        )
        current_domain.repository_for(Account).add(account)
        return str(account.id)

    @handle(MakeDeposit)
    def handle_make_deposit(self, command: MakeDeposit):
        repo = current_domain.repository_for(Account)
        account = repo.get(command.account_id)
        account.deposit(command.amount, reference=command.reference)
        repo.add(account)

    @handle(MakeWithdrawal)
    def handle_make_withdrawal(self, command: MakeWithdrawal):
        repo = current_domain.repository_for(Account)
        account = repo.get(command.account_id)
        account.withdraw(command.amount, reference=command.reference)
        repo.add(account)

    @handle(CloseAccount)
    def handle_close_account(self, command: CloseAccount):
        repo = current_domain.repository_for(Account)
        account = repo.get(command.account_id)
        account.close(reason=command.reason)
        repo.add(account)


domain.init(traverse=False)
domain.config["event_processing"] = "sync"
domain.config["command_processing"] = "sync"


if __name__ == "__main__":
    with domain.domain_context():
        # Open an account with $100
        account_id = domain.process(
            OpenAccount(
                account_number="ACC-001",
                holder_name="Alice Johnson",
                opening_deposit=100.00,
            )
        )
        print(f"Account opened: {account_id}")

        # Try to withdraw $200 — the invariant will catch this
        try:
            domain.process(MakeWithdrawal(account_id=account_id, amount=200.00))
        except ValidationError as e:
            print(f"Overdraft rejected: {e.messages}")

        # Try to close with a non-zero balance — invariant catches this too
        try:
            domain.process(
                CloseAccount(account_id=account_id, reason="Customer request")
            )
        except ValidationError as e:
            print(f"Close rejected: {e.messages}")

        # Withdraw all funds first, then close
        domain.process(MakeWithdrawal(account_id=account_id, amount=100.00))
        domain.process(CloseAccount(account_id=account_id, reason="Customer request"))

        repo = domain.repository_for(Account)
        account = repo.get(account_id)
        print(f"\nAccount status: {account.status}")
        print(f"Balance: ${account.balance:.2f}")

        assert account.status == "CLOSED"
        assert account.balance == 0.0
        print("\nAll checks passed!")

Next

Chapter 5: Testing the Ledger →