02.04.01 — The validate → assert_that → execute → raise_if_invalid flow#

Source file: src/ocarina/dsl/invariants/validate.py (entry-point) + src/ocarina/dsl/invariants/internals/validation_chain.py

The state machine#

Diagram#

validate(value: T)
       │
       ▼
ValidationStartBlock[T]
       │
       └─ assert_that(predicate, *, msg=None)
              │
              ▼
       ValidationAssertBlock[T]   ← state accepting several continuations
              │
              ├─► assert_that(P)     [loop, AND-chainable, see dedicated diagram]
              ├─► otherwise(P)       [OR, see 03-otherwise-any-of.md]
              ├─► then(U, name=...)  [value swap, see 04-then-chain-of-validations.md]
              └─► execute()          → _ValidationResult
                                          │
                                          ├─ is_valid: bool
                                          ├─ errors: Sequence[InvariantViolationError]
                                          ├─ validated_values: Sequence[Any]
                                          └─ raise_if_invalid() : None | raise AggregateInvariantViolationError

assert_that loop#

assert_that declares one assertion on the current value. Multiple assert_that calls chain — each adds another condition (logical AND, order preserved).

02.04.02 — Catalog of builtin assertions#

Source file: src/ocarina/dsl/invariants/assertions.py — ~25 predicates.

Every assertion follows the same contract:

• A direct predicate (value: T) -> None that raises InvariantViolationError on contract violation.
• Or a closure when you need a config argument (is_equal_to(cmp), etc.).
• Or, occasionally, a higher-order function (HOF) like each. Lives here pragmatically — a dedicated file for a single case would be overkill.

Full table#

A few implementations in detail#

is_email#

def is_email(value: str) -> None:
    """Assert that the string is a valid email address (fast check)."""
    if " " in value:
        raise InvariantViolationError(f"'{value}' must not contain whitespace.")
    if value.count("@") != 1:
        raise InvariantViolationError(f"'{value}' must contain exactly one '@' character.")
    local_part, domain_part = value.split("@")
    if not local_part or not domain_part:
        raise InvariantViolationError(f"'{value}' must have non-empty local and domain parts.")
    if "." not in domain_part:
        raise InvariantViolationError(f"Domain part of '{value}' must contain at least one '.'.")

Four bare-minimum checks. No RFC 5322 regex. Deliberate: real email validation happens by sending an email, not by regex.

02.04.03 — .otherwise(...) and _any_of#

How Ocarina expresses a logical OR between two predicates without breaking error aggregation.

The problem#

Concrete case: you want “age == 18 OR age <= 65.” The ROP/invariants syntax is a strict chain of assertions, and every .assert_that() is an AND. How do you express an OR?

Code#

def otherwise(self, fallback: Predicate[T], *, msg: str | None = None) -> ValidationAssertBlock[T]:
    if self._last_predicate is None:
        raise RuntimeError("otherwise() must follow assert_that().")

    fallback_with_msg = _with_msg(fallback, msg, self._name)
    combined = _any_of(
        self._last_predicate, *([*self._otherwise_predicates, fallback_with_msg])
    )
    self._chain._steps.pop()                    # replace the last step
    self._chain.add_assertion(self._value, combined, self._name)
    self._last_predicate = combined
    self._otherwise_predicates.append(fallback_with_msg)
    return self

1. Guard: otherwise() has to follow an assert_that(). With no prior predicate it raises. (In practice the type-checker already rules it out — ValidationStartBlock.otherwise doesn’t exist.)
2. Combination: combine previous predicate + all accumulated otherwise predicates + the new one, via _any_of. Result: a composite predicate.
3. Step replacement: pop the last step from _ValidationChain and push the combined one. Otherwise the original predicate AND the combined one would both be present, and the original error would resurface.

_any_of#

def _any_of[T](*predicates: _PredicateWithMsg[T]) -> _PredicateWithMsg[T]:
    def _try_predicate(p: _PredicateWithMsg[T], value: T) -> InvariantViolationError | None:
        try:
            p(value)
        except InvariantViolationError as exc:
            return exc
        else:
            return None

    def combined(value: T) -> None:
        errors = []
        for p in predicates:
            error = _try_predicate(p, value)
            if error is None:
                return                              # ✅ one predicate passes → all pass
            errors.append(error)

        formatted_error_messages = "  | " + "\n  | ".join(str(e) for e in errors)
        msg = (
            f"All predicates failed for value {value!r}.\n"
            "» At least one of the following conditions must be satisfied:\n"
            f"{formatted_error_messages}"
        )
        raise InvariantViolationError(msg)

    return _PredicateWithMsg(combined)

1. First success wins: the moment one predicate passes, combined returns None.
2. All fail: build an aggregated message listing every failure, separated by |.
3. Result is a _PredicateWithMsg: plug it back into the chain like any other predicate.
4. Aggregation is recursive: 3 successive otherwise() calls fold (P_orig OR P_otherwise1 OR P_otherwise2 OR P_otherwise3) into one predicate.

Example#

validate(age, name="age")
    .assert_that(is_equal_to(18), msg="Must be 18 or at most 65")
    .otherwise(is_less_than_or_equal_to(65), msg="Must be 18 or at most 65")
    .execute()
    .raise_if_invalid()

1. If age == 18: is_equal_to(18) passes → OK.
2. If age == 30: is_equal_to(18) raises, is_less_than_or_equal_to(65) passes → OK.
3. If age == 70: both raise → InvariantViolationError with:

age: All predicates failed for value 70.
» At least one of the following conditions must be satisfied:
  | age: Must be 18 or at most 65
  | age: Must be 18 or at most 65

(The duplicated message comes from passing the same msg= twice. Intentional here — we want a single statement of intent.)

02.04.04 — .then(...) and chain_validations(...)#

Two distinct mechanisms to compose multiple validations into one.

.then(new_value)#

def then[U](self, new_value: U, *, name: str | None = None) -> ValidationStartBlock[U]:
    return ValidationStartBlock(new_value, self._chain, name)

1. Switches the type from T to U. The checker accepts the change.
2. Keeps the chain (self._chain). Past assertions stay in it; new ones stack on top.
3. Swaps the name: the name you pass to .then() is the one used for the next assertions, not the previous block’s.

Canonical use case#

validate(user, name="user")
    .assert_that(is_not_none)
    .then(user.email, name="user.email")
    .assert_that(is_email)
    .then(user.age, name="user.age")
    .assert_that(is_positive)
    .assert_that(is_less_than_or_equal_to(120))
    .execute()
    .raise_if_invalid()

Reads as: validate user, then user.email, then user.age. All pass → OK. One fails → every error rolls up into the AggregateInvariantViolationError.

02.04.05 — BusinessInvariantValidator vs FrameworkInvariantValidator#

Two factories, identical code, exist purely to signal intent.

Code#

class _BaseCustomInvariantValidator:
    @staticmethod
    def create[T](
        value: T,
        name: str,
        build_chain: Callable[[ValidationStartBlock[T], T], ValidationAssertBlock[T]],
    ) -> ValidationAssertBlock[T]:
        return _create_business_invariant_validator(
            value, name, lambda chain: build_chain(chain, value)
        )


@final
class BusinessInvariantValidator(_BaseCustomInvariantValidator):
    """Factory for creating business domain invariant validators.

    Use this for domain-specific validation logic (e.g., "user must be adult",
    "order total must match items").
    """


@final
class FrameworkInvariantValidator(_BaseCustomInvariantValidator):
    """Factory for creating framework-level invariant validators.

    Use this for technical/framework validation logic (e.g., "config must be valid",
    "test structure must be correct").
    """

def _create_business_invariant_validator[T](
    value: T,
    name: str,
    build_chain: ValidationChainBuilder[T],
) -> ValidationAssertBlock[T]:
    start_block = validate(value, name=name)
    return build_chain(start_block)

Why two classes?#

1. BusinessInvariantValidator flags a business rule; FrameworkInvariantValidator, a technical framework rule. Intent jumps off the page.
2. Grep. FrameworkInvariantValidator.create finds every framework invariant in one search. Same for business rules on the project side.
3. Maintainability. Want to add behavior (different log, different stack trace) to one or the other? Do it without touching every call-site.

Usage#

Full pattern#

# src/ocarina/custom_invariants/testing/workers.py
from ocarina.dsl.invariants.assertions import is_not_zero, is_positive
from ocarina.dsl.invariants.validate import FrameworkInvariantValidator


def _workers_amount_chain(
    chain: ValidationStartBlock[int],
    value: int,
) -> ValidationAssertBlock[int]:
    msg = f"Value Error: Number of workers must be at least 1 (got: {value})."
    return chain.assert_that(is_positive, msg=msg).assert_that(is_not_zero, msg=msg)


def validate_workers_amount(
    *, workers_amount: int, name: str
) -> ValidationAssertBlock[int]:
    """Validate that workers amount is at least 1."""
    return FrameworkInvariantValidator.create(
        workers_amount, name, _workers_amount_chain
    )

1. Private _workers_amount_chain(chain, value) -> ValidationAssertBlock defines the what.
2. Public validate_workers_amount(*, workers_amount, name) is the API and delegates to FrameworkInvariantValidator.create.
3. Caller in TestSuite.run: validate_workers_amount(...).execute().raise_if_invalid().

Which factory for which case#

Holy Book (excerpt from the “Extensibility” chapter):

02.04.06 — Invariant error hierarchy#

Source file: src/ocarina/dsl/invariants/errors.py

InvariantViolationError (base — subclass of Exception)
├── DuplicatesError                  (raised by `has_unique_elements`)
└── AggregateInvariantViolationError (raised by `_ValidationResult.raise_if_invalid`)

InvariantViolationError#

class InvariantViolationError(Exception):
    """Base exception raised when an invariant is violated."""

Bare Exception subclass, no extra logic. Every predicate raises this (or a subclass).

try:
    validate(value).assert_that(predicate).execute().raise_if_invalid()
except InvariantViolationError as e:
    logger.error(f"Validation failed: {e}")

… catches every violation with a single clause.

DuplicatesError#

class DuplicatesError(InvariantViolationError):
    def __init__(self, duplicates: Sequence[Any], message: str | None = None) -> None:
        self.duplicates = duplicates
        msg = message or "Duplicate elements detected:\n" + "\n".join(
            f" - {d}" for d in duplicates
        )
        super().__init__(msg)

• Stashes the duplicate list in .duplicates.
• Default message: “Duplicate elements detected:\n - apple\n - banana”.
• Accepts a custom message=.

Raised by has_unique_elements.