Coverage for src / ocarina / dsl / invariants / internals / validation_chain.py: 25.40%
112 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-04 16:22 +0200
1"""Validation chain core.
3Architecture:
4 ValidationStartBlock: Entry point, awaits first assertion.
5 ValidationAssertBlock: Accepts more assertions, alternatives, or execution.
6 _ValidationChain: Internal accumulator of validation steps.
7 _ValidationResult: Result container with error aggregation.
9Type flow:
10 validate(value: T) ──> ValidationStartBlock[T]
11 |
12 └─ assert_that(predicate: Predicate[T])
13 |
14 v
15 ValidationAssertBlock[T]
16 |
17 ┌─────┴──────────────┐
18 | |
19 assert_that(P[T]) otherwise(P[T])
20 |
21 v
22 ValidationAssertBlock[T] (T preserved as long as value doesn't change)
23 |
24 └─ then(new_value: U)
25 |
26 v
27 ValidationStartBlock[U] (new type for the related value)
28 |
29 └─ assert_that(predicate: Predicate[U])
30 |
31 v
32 ValidationAssertBlock[U] (chain continues, type adapted to U)
34Key design decisions:
35 - Error aggregation: all steps run, errors collected rather than fail-fast.
36 - .otherwise() implements logical OR by combining predicates with _any_of().
37 - .then() threads the same _ValidationChain across value changes,
38 so chain_validations() and .execute() always see the full picture.
39 - _ValidationChain is mutable and shared across blocks in the same chain;
40 blocks hold a reference to it, not a copy.
41"""
43from collections.abc import Callable
44from typing import TYPE_CHECKING, Any, final
46from ocarina.dsl.invariants.errors import (
47 AggregateInvariantViolationError,
48 InvariantViolationError,
49)
51if TYPE_CHECKING:
52 from collections.abc import Sequence
54type Predicate[T] = Callable[[T], None]
57def _with_msg[T](
58 predicate: Predicate[T], msg: str | None, name: str | None = None
59) -> _PredicateWithMsg[T]:
60 """Wrap a predicate with an optional custom error message.
62 Args:
63 predicate: The validation function to wrap.
64 msg: Optional custom error message to use instead of predicate's default.
65 name: Optional name prefix for the error message (e.g., "email:").
67 Returns:
68 A wrapped predicate that uses the custom message on failure.
70 """
71 prefix = f"{name}:" if name else ""
72 full_msg = f"{prefix} {msg}" if msg else None
73 return _PredicateWithMsg(predicate, full_msg)
76@final
77class _PredicateWithMsg[T]:
78 """Internal wrapper that associates a predicate with a custom error message.
80 This allows predicates to have context-specific error messages while keeping
81 the predicate functions themselves reusable.
83 Attributes:
84 predicate: The validation function.
85 msg: Optional custom error message.
87 """
89 def __init__(self, predicate: Predicate[T], msg: str | None = None) -> None:
90 """Initialize the predicate wrapper.
92 Args:
93 predicate: The validation function to wrap.
94 msg: Optional custom error message to override predicate's default.
96 """
97 self.predicate = predicate
98 self.msg = msg
100 def __call__(self, value: T) -> None:
101 """Execute the predicate with custom error message handling.
103 Args:
104 value: The value to validate.
106 Raises:
107 InvariantViolationError: If validation fails.
109 """
110 try:
111 self.predicate(value)
112 except InvariantViolationError as exc:
113 if self.msg:
114 raise InvariantViolationError(self.msg) from exc
115 raise
118def _any_of[T](*predicates: _PredicateWithMsg[T]) -> _PredicateWithMsg[T]:
119 """Create a combined predicate that succeeds if ANY of the predicates succeed.
121 This implements logical OR for predicates, used by the .otherwise() method.
122 If all predicates fail, an error listing all failures is raised.
124 Args:
125 *predicates: Variable number of predicates to combine with OR logic.
127 Returns:
128 A single predicate that passes if at least one input predicate passes.
130 Example:
131 >>> # age must be exactly 18 OR less than or equal to 65
132 >>> combined = _any_of(is_equal_to(18), is_less_than_or_equal_to(65))
133 >>> combined(20) # Passes (satisfies second predicate)
134 >>> combined(70) # Raises (fails both predicates)
136 """
138 def _try_predicate(
139 p: _PredicateWithMsg[T], value: T
140 ) -> InvariantViolationError | None:
141 try:
142 p(value)
143 except InvariantViolationError as exc:
144 return exc
145 else:
146 return None
148 def combined(value: T) -> None:
149 errors = []
150 for p in predicates:
151 error = _try_predicate(p, value)
152 if error is None:
153 return
154 errors.append(error)
156 formatted_error_messages = " | " + "\n | ".join(str(e) for e in errors)
157 msg = (
158 f"All predicates failed for value {value!r}.\n"
159 "» At least one of the following conditions must be satisfied:\n"
160 f"{formatted_error_messages}"
161 )
162 raise InvariantViolationError(msg)
164 return _PredicateWithMsg(combined)
167@final
168class _ValidationResult:
169 """Container for validation results with error aggregation.
171 Attributes:
172 is_valid: True if all validations passed, False otherwise.
173 errors: List of all InvariantViolationErrors that occurred.
174 validated_values: List of values that passed validation.
176 """
178 def __init__(
179 self,
180 *,
181 is_valid: bool,
182 errors: Sequence[InvariantViolationError],
183 validated_values: Sequence[Any],
184 ) -> None:
185 """Initialize validation result.
187 Args:
188 is_valid: Whether all validations passed.
189 errors: Sequence of validation errors (empty if valid).
190 validated_values: Sequence of successfully validated values.
192 """
193 self.is_valid = is_valid
194 self.errors = errors
195 self.validated_values = validated_values
197 def raise_if_invalid(self) -> None:
198 """Raise an exception if validation failed.
200 Returns:
201 None.
203 Side Effects:
204 Raises an exception if validation failed.
206 Raises:
207 AggregateInvariantViolationError: If any validations failed,
208 containing all accumulated errors.
210 Example:
211 >>> result = validate(value).assert_that(predicate).execute()
212 >>> result.raise_if_invalid() # Raises if validation failed
214 """
215 if not self.is_valid:
216 raise AggregateInvariantViolationError(self.errors)
219@final
220class _ValidationChain:
221 """Internal accumulator for validation steps.
223 This class maintains the list of validation steps and executes them
224 sequentially, collecting errors rather than failing fast.
226 Attributes:
227 _steps: List of (value, name, predicate) tuples to execute.
229 """
231 def __init__(self) -> None:
232 """Initialize an empty validation chain."""
233 self._steps: list[tuple[Any, str | None, _PredicateWithMsg[Any]]] = []
235 def _merge_chain(self, other: _ValidationChain) -> None:
236 """Merge assertions from another chain into this one.
238 Args:
239 other: Another validation chain whose steps will be appended.
241 See Also:
242 - chain_validations
244 """
245 self._steps.extend(other._steps)
247 def add_assertion(
248 self,
249 value: Any, # noqa: ANN401
250 predicate: _PredicateWithMsg[Any],
251 name: str | None = None,
252 ) -> None:
253 """Add a validation step to the chain.
255 Args:
256 value: The value to validate.
257 predicate: The wrapped predicate to apply.
258 name: Optional name for better error messages.
260 """
261 self._steps.append((value, name, predicate))
263 def execute(self) -> _ValidationResult:
264 """Execute all validation steps and aggregate results.
266 Returns:
267 A ValidationResult containing success status, errors, and valid values.
269 Note:
270 This method does NOT raise exceptions. All errors are collected
271 and returned in the result. Use result.raise_if_invalid() to raise.
273 """
275 def run_step(
276 value: Any, # noqa: ANN401
277 predicate_with_msg: _PredicateWithMsg[Any],
278 ) -> tuple[bool, Any | InvariantViolationError]:
279 try:
280 predicate_with_msg(value)
281 except InvariantViolationError as exc:
282 return False, exc
283 else:
284 return True, value
286 errors = []
287 validated = []
288 for value, _, predicate_with_msg in self._steps:
289 success, outcome = run_step(value, predicate_with_msg)
290 if success:
291 validated.append(outcome)
292 else:
293 errors.append(outcome)
294 return _ValidationResult(
295 is_valid=(len(errors) == 0), errors=errors, validated_values=validated
296 )
299@final
300class ValidationStartBlock[T]:
301 """Entry point for a validation chain, awaiting the first assertion.
303 This is the initial state returned by validate().
304 It provides only the assert_that() method to begin validation.
306 Type Parameters:
307 T: The type of value being validated.
309 Example:
310 >>> start = validate(42, name="age")
311 >>> start.assert_that(is_positive) # Returns ValidationAssertBlock
313 """
315 def __init__(
316 self, value: T, chain: _ValidationChain | None = None, name: str | None = None
317 ) -> None:
318 """Initialize a validation start block.
320 Args:
321 value: The value to validate.
322 chain: Optional existing chain to continue (used internally by .then()).
323 name: Optional name for this value in error messages.
325 """
326 self._value = value
327 self._chain = chain or _ValidationChain()
328 self._name = name
330 def assert_that(
331 self, predicate: Predicate[T], *, msg: str | None = None
332 ) -> ValidationAssertBlock[T]:
333 """Add the first validation predicate to the chain.
335 Args:
336 predicate: A validation function that raises on failure.
337 msg: Optional custom error message to use instead of predicate's default.
339 Returns:
340 A ValidationAssertBlock for further chaining.
342 Example:
343 >>> validate(email, name="email")
344 ... .assert_that(is_str)
345 ... .assert_that(is_email)
347 """
348 predicate = _with_msg(predicate, msg, self._name)
349 self._chain.add_assertion(self._value, predicate, self._name)
350 return ValidationAssertBlock(
351 self._value, self._chain, self._name, last_predicate=predicate
352 )
355@final
356class ValidationAssertBlock[T]:
357 """Validation block after at least one assertion has been added.
359 This block provides multiple options:
360 - .assert_that(): Add another validation
361 - .otherwise(): Add an alternative predicate (logical OR)
362 - .then(): Switch to validating a related value
363 - .execute(): Run all validations and get results
365 Type Parameters:
366 T: The type of value being validated.
368 Example:
369 >>> validate(age)
370 ... .assert_that(is_positive)
371 ... .assert_that(is_less_than_or_equal_to(120))
372 ... .execute()
373 ... .raise_if_invalid()
375 """
377 def __init__(
378 self,
379 value: T,
380 chain: _ValidationChain,
381 name: str | None = None,
382 last_predicate: _PredicateWithMsg[T] | None = None,
383 ) -> None:
384 """Initialize a validation assert block.
386 Args:
387 value: The value being validated.
388 chain: The validation chain accumulating steps.
389 name: Optional name for this value in error messages.
390 last_predicate: The most recently added predicate (for .otherwise()).
392 """
393 self._value = value
394 self._chain = chain
395 self._name = name
396 self._last_predicate = last_predicate
397 self._otherwise_predicates: list[_PredicateWithMsg[T]] = []
399 def assert_that(
400 self, predicate: Predicate[T], msg: str | None = None
401 ) -> ValidationAssertBlock[T]:
402 """Add another validation predicate (logical AND).
404 Args:
405 predicate: A validation function that raises on failure.
406 msg: Optional custom error message.
408 Returns:
409 Self for chaining.
411 Example:
412 >>> validate(password)
413 ... .assert_that(has_min_length(8))
414 ... .assert_that(contains_uppercase)
415 ... .assert_that(contains_digit)
417 """
418 predicate = _with_msg(predicate, msg, self._name)
419 self._chain.add_assertion(self._value, predicate, self._name)
420 self._last_predicate = predicate
421 self._otherwise_predicates = []
422 return self
424 def otherwise(
425 self, fallback: Predicate[T], *, msg: str | None = None
426 ) -> ValidationAssertBlock[T]:
427 """Add an alternative predicate.
429 Logical OR with the last assertion and any previously added alternatives.
431 The validation passes if EITHER the last assert_that() OR this otherwise()
432 succeeds. Multiple otherwise() calls create a multi-way OR.
434 Args:
435 fallback: Alternative validation predicate.
436 msg: Optional custom error message for this alternative.
438 Returns:
439 Self for chaining.
441 Example:
442 >>> validate(age)
443 ... .assert_that(is_equal_to(18), msg="Must be 18")
444 ... .otherwise(is_equal_to(21), msg="Or must be 21")
445 ... .otherwise(is_equal_to(65), msg="Or must be 65")
447 """
448 if self._last_predicate is None: # pragma: no cover
449 """Note: Guard to make the type-system happy:
450 this case isn't allowed by the fluent API."""
452 msg = "otherwise() must follow assert_that()."
453 raise RuntimeError(msg)
455 fallback_with_msg = _with_msg(fallback, msg, self._name)
456 combined = _any_of(
457 self._last_predicate, *([*self._otherwise_predicates, fallback_with_msg])
458 )
459 self._chain._steps.pop() # noqa: SLF001
460 self._chain.add_assertion(self._value, combined, self._name)
461 self._last_predicate = combined
462 self._otherwise_predicates.append(fallback_with_msg)
463 return self
465 def then[U](
466 self, new_value: U, *, name: str | None = None
467 ) -> ValidationStartBlock[U]:
468 """Switch to validating a related value, continuing the same chain.
470 This allows validating multiple related values in sequence, with all
471 errors aggregated together.
473 Args:
474 new_value: The next value to validate.
475 name: Optional name for the new value in error messages.
477 Returns:
478 A new ValidationStartBlock for the new value.
480 Example:
481 >>> validate(user, name="user")
482 ... .assert_that(is_not_none)
483 ... .then(user.email, name="email")
484 ... .assert_that(is_email)
485 ... .then(user.age, name="age")
486 ... .assert_that(is_positive)
487 ... .execute()
489 """
490 return ValidationStartBlock(new_value, self._chain, name)
492 def execute(self) -> _ValidationResult:
493 """Execute all validation steps and return aggregated results.
495 Returns:
496 A ValidationResult containing all errors and validated values.
498 Note:
499 This does NOT raise exceptions. Use .raise_if_invalid() on the
500 result to raise an exception if validation failed.
502 Example:
503 >>> result = validate(x).assert_that(predicate).execute()
504 >>> if not result.is_valid:
505 ... logger.error(f"Validation failed: {result.errors}")
507 """
508 return self._chain.execute()
511def chain_validations(
512 first: ValidationAssertBlock[Any],
513 *rest: ValidationAssertBlock[Any],
514) -> ValidationAssertBlock[Any]:
515 """Merge multiple independent validation chains into one.
517 This allows combining pre-built validation chains, useful for composing
518 complex validations from reusable pieces.
520 Args:
521 first: The first validation chain.
522 *rest: Additional validation chains to merge.
524 Returns:
525 A single ValidationAssertBlock containing all merged validations.
527 Example:
528 >>> user_validation = validate(user).assert_that(is_not_none)
529 >>> email_validation = validate(email).assert_that(is_email)
530 >>> combined = chain_validations(user_validation, email_validation)
531 >>> combined.execute().raise_if_invalid()
533 """
534 merged_chain = _ValidationChain()
535 merged_chain._merge_chain(first._chain) # noqa: SLF001
537 for block in rest:
538 merged_chain._merge_chain(block._chain) # noqa: SLF001
540 return ValidationAssertBlock(first._value, merged_chain, first._name) # noqa: SLF001