Invariants

DDD CQRS ES

Field-level validations catch type errors, missing values, and range violations, but they cannot express rules that span multiple fields or depend on the aggregate's overall state. For example, no single field constraint can enforce business rules such as "an order's total must equal the sum of its items" or "a shipment can only be dispatched if the order is confirmed".

Invariants fill this gap. They are business rules or constraints that must always be true within a domain concept. Protean treats invariants as explicit and visible in the model, with the @invariant decorator. You can define invariants on Aggregates, Entities, Value Objects, and Domain Services.

For background on why invariants are fundamental to DDD and how they keep your domain always valid, see Invariants concept.

@invariant decorator

Invariants are defined using the @invariant decorator with either a .pre or .post qualifier. You must always use @invariant.pre or @invariant.post, plain @invariant without a qualifier is not valid.

from datetime import date, timedelta
from enum import Enum

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

domain = Domain()


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


@domain.aggregate
class Order:
    customer_id: Identifier()
    order_date: Date()
    total_amount: Float()
    status: String(max_length=50, choices=OrderStatus)
    items = HasMany("OrderItem")

    @invariant.post
    def total_amount_of_order_must_equal_sum_of_subtotal_of_all_items(self):
        if self.total_amount != sum(item.subtotal for item in self.items):
            raise ValidationError({"_entity": ["Total should be sum of item prices"]})

    @invariant.post
    def order_date_must_be_within_the_last_30_days_if_status_is_pending(self):
        if (
            self.status == OrderStatus.PENDING.value
            and not date.today() - timedelta(days=30) <= self.order_date <= date.today()
        ):
            raise ValidationError(
                {
                    "_entity": [
                        "Order date must be within the last 30 days if status is PENDING"
                    ]
                }
            )

    @invariant.post
    def customer_id_must_be_non_null_and_the_order_must_contain_at_least_one_item(self):
        if not self.customer_id or not self.items:
            raise ValidationError(
                {
                    "_entity": [
                        "Customer ID must be non-null and the order must contain at least one item"
                    ]
                }
            )


@domain.entity(part_of=Order)
class OrderItem:
    product_id: Identifier()
    quantity: Integer()
    price: Float()
    subtotal: Float()

    class Meta:
        part_of = Order

    @invariant.post
    def the_quantity_must_be_a_positive_integer_and_the_subtotal_must_be_correctly_calculated(
        self,
    ):
        if self.quantity <= 0 or self.subtotal != self.quantity * self.price:
            raise ValidationError(
                {
                    "_entity": [
                        "Quantity must be a positive integer and the subtotal must be correctly calculated"
                    ]
                }
            )

In the above example, the Order aggregate has three invariants (business conditions). The total amount of the order must always equal the sum of individual item subtotals. The order date must be within the last 30 days if the status is PENDING. The order must have a customer and at least one item. The OrderItem entity has an invariant of its own.

All methods marked @invariant are associated with the domain element when the element is registered with the domain.

Invariant methods must raise ValidationError (from protean.exceptions) when the rule is violated. This is the exception type the framework catches and aggregates. If an invariant raises a different exception type, it will propagate directly to the caller instead of being collected with other invariant violations.

pre and post Invariants

The @invariant decorator has two flavors, pre and post.

post invariants are triggered after elements are constructed or updated. They ensure that the aggregate is in a valid state after the change.

pre invariants are triggered before elements are updated. They are used to check whether a proposed change is allowed given the current state.

Pre-invariant example

Pre-invariants are useful when you want to guard against invalid transitions. For example, checking that an account has sufficient balance before a withdrawal:

from protean import Domain, invariant
from protean.exceptions import ValidationError
from protean.fields import Float, Identifier, String

domain = Domain()


@domain.aggregate
class Account:
    account_number: Identifier(required=True, unique=True)
    balance: Float(default=0.0)
    status: String(choices=["ACTIVE", "FROZEN"], default="ACTIVE")

    @invariant.pre
    def account_must_be_active_to_transact(self):
        if self.status == "FROZEN":
            raise ValidationError({"_entity": ["Cannot modify a frozen account"]})

    @invariant.post
    def balance_must_not_be_negative(self):
        if self.balance < 0:
            raise ValidationError({"_entity": ["Insufficient funds"]})

    def withdraw(self, amount: float):
        self.balance -= amount

When withdraw() is called, the flow is:

1. Pre-invariants fire, account_must_be_active_to_transact checks the current state. If the account is frozen, ValidationError is raised and the assignment self.balance -= amount never happens.
2. The attribute assignment self.balance -= amount executes.
3. Post-invariants fire, balance_must_not_be_negative checks the resulting state. If the balance went negative, ValidationError is raised and the assignment is rolled back: balance keeps its previous value and the account's changed flag goes back to what it was before the assignment.

The rollback undoes only the assignment that failed. If a method changes two fields and the second change breaks a post-invariant, the first change stays.

Rollback applies to every kind of field, on the aggregate and on its child entities: plain fields, ValueObject fields with their embedded attributes, Reference fields with their <name>_id attribute, and HasOne and HasMany fields. A failed add_<field>() or remove_<field>() call is undone the same way, including the removal of the child's own children. The assignment is also undone when an invariant fails with an error other than ValidationError, such as a TypeError from comparing None.

A change made to a List or Dict field in place, such as self.tags.append("x") or self.tags += ["x"], alters the existing value before Protean sees an assignment, so a failed check cannot undo it. To make the change undoable, assign a new value: self.tags = [*self.tags, "x"]. Inside an atomic_change block, in-place changes are undone as well, because the block keeps a copy of every field from block entry.

An atomic_change block is rolled back as a whole: when the check at its end fails, the aggregate and its child entities go back to their state from block entry, including association changes (see below). Events raised inside the block are discarded too. An exception raised by the code inside the block does not roll it back. On an event-sourced aggregate, raise_() runs the @apply handler inside atomic_change. So when a check fails after the handler runs, raise_() undoes the handler's field changes and also discards the event.

Note

pre invariants are not applicable when aggregates and entities are being initialized. Their validations only kick in when an element is being changed or updated from an existing state.

Note

pre invariant checks are not applicable to ValueObject elements because they are immutable. They cannot be changed once initialized.

When Invariants Run

Invariant validations are triggered throughout the lifecycle of domain objects. The aggregate is the root of the triggering mechanism. Protean runs validations recursively: first on the aggregate, then on each enclosed entity.

Post-Initialization

Immediately after an object (aggregate or entity) is initialized, all post-invariant checks are triggered to ensure the aggregate starts in a valid state.

In [1]: from datetime import date

In [2]: Order(
   ...:    customer_id="1",
   ...:    order_date=date.today(),
   ...:    total_amount=100.0,
   ...:    status="PENDING",
   ...:    items=[
   ...:        OrderItem(product_id="1", quantity=2, price=10.0, subtotal=20.0),
   ...:        OrderItem(product_id="2", quantity=3, price=20.0, subtotal=60.0),
   ...:    ],
   ...:)
ERROR: Error during initialization: {'_entity': ['Total should be sum of item prices']}
...
ValidationError: {'_entity': ['Total should be sum of item prices']}

Attribute Changes

Every attribute change in an aggregate or its enclosed entities triggers invariant validation throughout the aggregate cluster. As a result, any modification keeps the domain model consistent.

In [1]: from datetime import date

In [2]: order = Order(
   ...:     customer_id="1",
   ...:     order_date=date.today(),
   ...:     total_amount=100.0,
   ...:     status="PENDING",
   ...:     items=[
   ...:         OrderItem(product_id="1", quantity=4, price=10.0, subtotal=40.0),
   ...:         OrderItem(product_id="2", quantity=3, price=20.0, subtotal=60.0),
   ...:     ],
   ...: )
   ...:

In [3]: order.total_amount = 140.0
...
ValidationError: {'_entity': ['Total should be sum of item prices']}

Validation runs on every assignment because __setattr__ intercepts every field assignment and runs the full pre-check → validate → post-check cycle. See Aggregate Mutation for the complete mechanism.

Entity-to-Root Delegation

When a child entity's attribute is changed, invariants fire on the root aggregate, not just the entity itself. This ensures that cross-entity business rules (defined on the aggregate) are always enforced, even when the mutation happens deep in the aggregate cluster.

Adding and Removing Entities

add_* and remove_* methods on HasMany associations also trigger invariants. Both pre-invariants and post-invariants fire, just as they do for direct attribute assignments.

In [4]: order.add_items(OrderItem(product_id="3", quantity=2, price=10.0, subtotal=20.0))
...
ValidationError: {'_entity': ['Total should be sum of item prices']}

Adding an item changes the sum of item subtotals. The new sum no longer matches total_amount, so the post-invariant fires immediately.

When adding or removing entities requires coordinated changes (like updating a total), use atomic_change to batch the mutations.

Atomic Changes

Sometimes you need to change multiple attributes together. In that case, validations should not trigger until the entire operation is complete. The atomic_change context manager can be used to achieve this.

Within the atomic_change context manager, the cycle works as follows:

1. Pre-invariants fire on entry: The current state is validated.
2. Invariant checks are suspended during the block. Individual assignments do not trigger pre/post checks.
3. Post-invariants fire on exit: The final state is validated. If a check fails, ValidationError is raised and every change made inside the block is undone.

from protean import atomic_change


def add_third_item(order):
    with atomic_change(order):
        order.total_amount = 120.0
        order.add_items(
            OrderItem(product_id="3", quantity=2, price=10.0, subtotal=20.0)
        )

Called on the order from the earlier examples, add_third_item() changes the total and adds the item, and the check at the end of the block passes:

In [1]: from datetime import date

In [2]: order = Order(
   ...:     customer_id="1",
   ...:     order_date=date.today(),
   ...:     total_amount=100.0,
   ...:     status="PENDING",
   ...:     items=[
   ...:         OrderItem(product_id="1", quantity=4, price=10.0, subtotal=40.0),
   ...:         OrderItem(product_id="2", quantity=3, price=20.0, subtotal=60.0),
   ...:     ],
   ...: )

In [3]: add_third_item(order)

In [4]: order.total_amount, len(order.items)
Out[4]: (120.0, 3)

If you make the same two updates one after another, outside atomic_change, the first update fails. On a fresh order with the same two items, setting the total alone breaks the post-invariant:

In [5]: order = Order(
   ...:     customer_id="1",
   ...:     order_date=date.today(),
   ...:     total_amount=100.0,
   ...:     status="PENDING",
   ...:     items=[
   ...:         OrderItem(product_id="1", quantity=4, price=10.0, subtotal=40.0),
   ...:         OrderItem(product_id="2", quantity=3, price=20.0, subtotal=60.0),
   ...:     ],
   ...: )

In [6]: order.total_amount = 120.0
...
ValidationError: {'_entity': ['Total should be sum of item prices']}

atomic_change is also used internally by raise_() in event-sourced aggregates to wrap @apply handlers, see Raising Events for details.

Note

atomic_change can only be applied when updating or changing an already initialized element.

Invariant Inheritance

Invariants defined on a parent class are inherited by subclasses through standard Python MRO resolution. A subclass inherits all @invariant.pre and @invariant.post methods from its parents, and can add its own.

Error Structure

When invariants raise ValidationError, the framework aggregates all violations into a single ValidationError with a dictionary structure:

ValidationError({
    '_entity': ['Total should be sum of item prices'],
    'balance': ['Insufficient funds'],
})

The _entity key is a convention for errors that apply to the entity as a whole rather than a specific field. Multiple invariant violations are collected and reported together.

Coded diagnostics

The raised ValidationError also carries a stable code you can act on without reading the messages. On an aggregate, entity, or domain service, a failed @invariant.post carries INVARIANT_POST_FAILED and a failed @invariant.pre carries INVARIANT_PRE_FAILED; a failed value object invariant carries VALUE_OBJECT_INVARIANT_FAILED. The exception's rationale and fix resolve from that code.

def change_total(order):
    try:
        order.total_amount = 140.0
    except ValidationError as exc:
        print(exc.code)  # "INVARIANT_POST_FAILED"
        print(exc.codes)  # ["INVARIANT_POST_FAILED"]
        print(exc.fix)  # how to fix it, from the registry

When several invariants fail together, exc.codes lists each distinct code once. Two failed @invariant.post checks give ["INVARIANT_POST_FAILED"], so exc.code is still "INVARIANT_POST_FAILED". When the failures carry different codes, exc.code is None because no single one names the failure. The errors dict lists every message either way.

Pass code to stamp a specific code on an invariant:

# fragment
@invariant.post(code="ORDER_TOTAL_MISMATCH")
def total_matches_items(self):
    if self.total_amount != sum(item.subtotal for item in self.items):
        raise ValidationError({"_entity": ["Total should be sum of item prices"]})

The failure then carries "ORDER_TOTAL_MISMATCH" in code and codes. A code you define yourself is carried as given; rationale and fix resolve only for codes in Protean's registry. See the init-time and runtime diagnostics reference for the built-in codes.

---

See also

Background: The Always-Valid Domain: The complete story of how Protean's four validation layers work together to guarantee your domain objects are never invalid.

Concept overview: Invariants: Why invariants are fundamental to DDD and how they keep your domain always valid.

Related guides:

• Validations: Field-level constraints (Layer 1).
• Status Transitions: Enforcing state machine rules with the Status field.
• Aggregate Mutation: The __setattr__ mechanism that triggers invariants.
• Domain Services: Cross-aggregate invariants in domain services.

Patterns: Validation Layering: Choosing the right layer for each kind of validation rule.