Chapter 5: Business Rules and Invariants

A domain model without business rules is just a data container. In this chapter we add invariants — rules that keep our aggregates in a valid state — and encapsulate behavior in aggregate methods.

Field-Level Validation

We have already seen some validation through field options:

• required=True — the field must be present
• choices=OrderStatus — the value must be one of the enum members
• max_length=150 — the string cannot exceed this length

When validation fails, Protean raises a ValidationError:

>>> Order(customer_name="", status="INVALID")
ValidationError: {
    'customer_name': ['is required'],
    'status': ["Value 'INVALID' is not a valid choice. ..."]
}

Field validation catches simple data errors at the boundary. But some rules involve multiple fields or depend on the aggregate's state. That is where invariants come in.

Post-Invariants: Validating State

A post-invariant is checked after every state change — on creation and on every subsequent mutation. Use @invariant.post:

    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']}

When Post-Invariants Run

Post-invariants run after initialization and after every attribute change. They guarantee that the aggregate is always in a valid state. If a mutation violates an invariant, the change is rejected and the aggregate stays in its previous state.

Pre-Invariants: Guarding Transitions

A pre-invariant is checked before state changes are applied. 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.add_item("Sapiens", 1, Money(amount=18.99))
ValidationError: {'_entity': ['Cannot modify an order that has been shipped']}

Pre vs Post: When to Use Which

Use Pre-Invariants When...	Use Post-Invariants When...
Guarding state transitions	Validating the resulting state
Preventing invalid operations	Ensuring multi-field consistency
Checking "can this happen?"	Checking "is this state valid?"

Aggregate Methods

Rather than mutating aggregate fields directly, 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 your aggregate. External code calls order.confirm() rather than order.status = "CONFIRMED". This keeps business logic inside the aggregate where invariants can enforce it.

Atomic Changes

Sometimes you need to make multiple changes that would individually violate invariants but result in a valid state. Use atomic_change to temporarily defer invariant checks:

from protean import atomic_change

with atomic_change(order) as order:
    order.status = "PROCESSING"  # Intermediate state
    order.customer_name = "Updated Name"
    # Invariants are deferred until the block exits
# Invariants checked here — if final state is valid, it passes

Inside the atomic_change block, invariants are suspended. They run when the block exits — if the final state is invalid, the error is raised then.

Error Handling Patterns

ValidationError carries a dictionary of error messages:

from protean.exceptions import ValidationError

try:
    Order(customer_name="")
except ValidationError as e:
    print(e.messages)
    # {'customer_name': ['is required'],
    #  '_entity': ['An order must contain at least one item']}

• Field-level errors are keyed by field name: {'customer_name': [...]}
• Entity-level errors use the _entity key: {'_entity': [...]}
• Service-level errors use _service (seen in Chapter 10)

This structure makes it straightforward to present errors in a UI — map field errors to form inputs, and entity errors to a general message area.

Putting It Together

    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

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!

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!")

Summary

In this chapter you learned:

• Field validation (required, choices, max_length) catches basic data errors at creation time.
• Post-invariants (@invariant.post) validate state after every change, keeping the aggregate always consistent.
• Pre-invariants (@invariant.pre) guard state transitions, preventing invalid operations.
• Aggregate methods encapsulate business logic and serve as the public API for state changes.
• atomic_change lets you make multi-step changes without triggering intermediate invariant checks.

We now have a rich domain model with aggregates, entities, value objects, and business rules. In the next part, we will add commands and events — the building blocks of an event-driven architecture.

Next

Chapter 6: Commands and Command Handlers →