Chapter 4: Business Rules and Invariants

In this chapter we will add invariants to our Order aggregate so that orders cannot be empty, and shipped orders cannot be modified.

Post-Invariants: Validating State

Let's add a rule: every order must have at least one item. A post-invariant is checked after every state change — on creation and on every subsequent mutation:

@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 add_item(self, book_title: str, quantity: int, unit_price: Money):
        """Add an item to this order."""
        self.add_items(
            OrderItem(
                book_title=book_title,
                quantity=quantity,
                unit_price=unit_price,
            )
        )

    def confirm(self):
        """Confirm the order for processing."""
        self.status = OrderStatus.CONFIRMED.value

    def ship(self):
        """Mark the order as shipped."""
        self.status = OrderStatus.SHIPPED.value

    @invariant.post
    def order_must_have_items(self):
        if not self.items or len(self.items) == 0:
            raise ValidationError(
                {"_entity": ["An order must contain at least one item"]}
            )

    @invariant.post
    def confirmed_order_must_have_multiple_items_or_quantity(self):
        if self.status == OrderStatus.CONFIRMED.value:
            total_quantity = sum(item.quantity for item in self.items)
            if total_quantity < 1:
                raise ValidationError(
                    {"_entity": ["A confirmed order must have at least 1 item"]}
                )

    @invariant.pre
    def cannot_modify_shipped_order(self):
        if self.status == OrderStatus.SHIPPED.value:
            raise ValidationError(
                {"_entity": ["Cannot modify an order that has been shipped"]}
            )

The order_must_have_items invariant runs whenever the Order is created or modified. If the items list is empty, it rejects the change:

>>> Order(customer_name="Alice")
ValidationError: {'_entity': ['An order must contain at least one item']}

Notice that the invariant runs automatically — we don't call it ourselves. Protean guarantees that the aggregate is always in a valid state.

Pre-Invariants: Guarding Transitions

A pre-invariant is checked before state changes are applied. We use it to prevent invalid operations:

@invariant.pre
def cannot_modify_shipped_order(self):
    if self.status == OrderStatus.SHIPPED.value:
        raise ValidationError(
            {"_entity": ["Cannot modify an order that has been shipped"]}
        )

This prevents any modifications once an order has been shipped:

>>> order.ship()
>>> order.customer_name = "Bob"
ValidationError: {'_entity': ['Cannot modify an order that has been shipped']}

Aggregate Methods

Rather than mutating aggregate fields directly, we encapsulate behavior in methods:

def add_item(self, book_title: str, quantity: int, unit_price: Money):
    """Add an item to this order."""
    self.add_items(
        OrderItem(
            book_title=book_title,
            quantity=quantity,
            unit_price=unit_price,
        )
    )

def confirm(self):
    """Confirm the order for processing."""
    self.status = OrderStatus.CONFIRMED.value

def ship(self):
    """Mark the order as shipped."""
    self.status = OrderStatus.SHIPPED.value

Methods become the public API of the aggregate. External code calls order.confirm() rather than order.status = "CONFIRMED". This keeps business logic inside the aggregate where invariants can enforce it.

Putting It Together

if __name__ == "__main__":
    with domain.domain_context():
        repo = domain.repository_for(Order)

        # --- Field-level validation ---
        print("=== Field Validation ===")
        try:
            Order(customer_name="")  # required field is empty
        except ValidationError as e:
            print(f"Caught: {e.messages}")

        try:
            Order(customer_name="Alice", status="INVALID_STATUS")
        except ValidationError as e:
            print(f"Caught: {e.messages}")

        # --- Invariant: order must have items ---
        print("\n=== Post-Invariant: Must Have Items ===")
        try:
            Order(customer_name="Alice")
        except ValidationError as e:
            print(f"Caught: {e.messages}")

        # --- Using aggregate methods ---
        print("\n=== Aggregate Methods ===")
        order = Order(
            customer_name="Alice",
            items=[
                OrderItem(
                    book_title="The Great Gatsby",
                    quantity=1,
                    unit_price=Money(amount=12.99),
                ),
            ],
        )
        # Add more items using the aggregate method
        order.add_item("Brave New World", 2, Money(amount=14.99))

        print(f"Order: {order.customer_name}, {len(order.items)} items")
        print(f"Status: {order.status}")

        # Confirm the order
        order.confirm()
        print(f"After confirm: {order.status}")

        # Ship the order
        order.ship()
        print(f"After ship: {order.status}")

        # --- Pre-Invariant: cannot modify shipped order ---
        print("\n=== Pre-Invariant: Cannot Modify Shipped ===")
        try:
            order.customer_name = "Bob"  # Any mutation triggers pre-check
        except ValidationError as e:
            print(f"Caught: {e.messages}")

        print("\nAll checks passed!")

Run it:

$ python bookshelf.py
=== Field Validation ===
Caught: {'customer_name': ['is required']}
Caught: {'status': ["Value 'INVALID_STATUS' is not a valid choice. ..."]}

=== Post-Invariant: Must Have Items ===
Caught: {'_entity': ['An order must contain at least one item']}

=== Aggregate Methods ===
Order: Alice, 2 items
Status: PENDING
After confirm: CONFIRMED
After ship: SHIPPED

=== Pre-Invariant: Cannot Modify Shipped ===
Caught: {'_entity': ['Cannot modify an order that has been shipped']}

All checks passed!

Notice how every rule is enforced automatically — the aggregate never enters an invalid state.

What We Built

• Post-invariants (@invariant.post) that validate state after every change, keeping the aggregate always consistent.
• Pre-invariants (@invariant.pre) that guard state transitions, preventing invalid operations.
• Aggregate methods that encapsulate business logic and serve as the public API for state changes.

We now have a rich domain model with aggregates, entities, value objects, and business rules. In the next chapter, we will add commands and handlers — the entry point for all state changes.

Full Source

from enum import Enum

from protean import Domain, invariant
from protean.exceptions import ValidationError
from protean.fields import Float, HasMany, Integer, String, ValueObject

domain = Domain()


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


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


@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 add_item(self, book_title: str, quantity: int, unit_price: Money):
        """Add an item to this order."""
        self.add_items(
            OrderItem(
                book_title=book_title,
                quantity=quantity,
                unit_price=unit_price,
            )
        )

    def confirm(self):
        """Confirm the order for processing."""
        self.status = OrderStatus.CONFIRMED.value

    def ship(self):
        """Mark the order as shipped."""
        self.status = OrderStatus.SHIPPED.value

    @invariant.post
    def order_must_have_items(self):
        if not self.items or len(self.items) == 0:
            raise ValidationError(
                {"_entity": ["An order must contain at least one item"]}
            )

    @invariant.post
    def confirmed_order_must_have_multiple_items_or_quantity(self):
        if self.status == OrderStatus.CONFIRMED.value:
            total_quantity = sum(item.quantity for item in self.items)
            if total_quantity < 1:
                raise ValidationError(
                    {"_entity": ["A confirmed order must have at least 1 item"]}
                )

    @invariant.pre
    def cannot_modify_shipped_order(self):
        if self.status == OrderStatus.SHIPPED.value:
            raise ValidationError(
                {"_entity": ["Cannot modify an order that has been shipped"]}
            )




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


domain.init(traverse=False)


if __name__ == "__main__":
    with domain.domain_context():
        repo = domain.repository_for(Order)

        # --- Field-level validation ---
        print("=== Field Validation ===")
        try:
            Order(customer_name="")  # required field is empty
        except ValidationError as e:
            print(f"Caught: {e.messages}")

        try:
            Order(customer_name="Alice", status="INVALID_STATUS")
        except ValidationError as e:
            print(f"Caught: {e.messages}")

        # --- Invariant: order must have items ---
        print("\n=== Post-Invariant: Must Have Items ===")
        try:
            Order(customer_name="Alice")
        except ValidationError as e:
            print(f"Caught: {e.messages}")

        # --- Using aggregate methods ---
        print("\n=== Aggregate Methods ===")
        order = Order(
            customer_name="Alice",
            items=[
                OrderItem(
                    book_title="The Great Gatsby",
                    quantity=1,
                    unit_price=Money(amount=12.99),
                ),
            ],
        )
        # Add more items using the aggregate method
        order.add_item("Brave New World", 2, Money(amount=14.99))

        print(f"Order: {order.customer_name}, {len(order.items)} items")
        print(f"Status: {order.status}")

        # Confirm the order
        order.confirm()
        print(f"After confirm: {order.status}")

        # Ship the order
        order.ship()
        print(f"After ship: {order.status}")

        # --- Pre-Invariant: cannot modify shipped order ---
        print("\n=== Pre-Invariant: Cannot Modify Shipped ===")
        try:
            order.customer_name = "Bob"  # Any mutation triggers pre-check
        except ValidationError as e:
            print(f"Caught: {e.messages}")

        print("\nAll checks passed!")

Next

Chapter 5: Commands and Handlers →