Validation

An external call can succeed and still return something you cannot use: an empty completion, a truncated document, a payload missing required fields. Validation turns such a value into an ordinary attempt failure, so retries and fallbacks can recover from it the same way they recover from exceptions.

Validators and policies#

A validator checks one value and returns a ValidationResult. A validation policy runs one or more validators in order.

from relprim import resilient, validation_policy, validator

non_empty = validator(
    "non_empty_response",
    lambda value: bool(value.strip()),
    message="Response must not be empty.",
)
not_truncated = validator(
    "complete_response",
    lambda value: not value.endswith("..."),
    message="Response looks truncated.",
)


@resilient(retries=2, validation=validation_policy(non_empty, not_truncated))
async def generate(prompt: str) -> str:
    return await provider.generate(prompt)

validator(name, predicate, *, message) creates a dependency-free CallableValidator: the predicate returns True for acceptable values, and message becomes the failure reason. Names and messages must not be blank.

Policies are fail-fast. Validators run in order and the first failure stops the policy. In the builder API, use .with_validation(policy).

Results#

result = non_empty.validate("   ")

result.valid            # False
result.validator_name   # "non_empty_response"
result.reason           # "Response must not be empty."

ValidationPolicy.validate(value) returns the first failing result or, when every validator passed, the last validator’s success result, so the reported name always belongs to a validator that ran. validate_or_raise(value) returns that success result or raises ValidationFailedError.

Failure behavior#

Inside an operation, the policy runs after each successful call, within the attempt and inside the circuit breaker, but outside the timeout. A failed result raises ValidationFailedError (with .validator_name and .reason) for that attempt, and then:

• it is retried when it matches retry_on. The default (Exception,) matches it. To retry only validation failures, or to exclude them, set retry_on explicitly:

  from relprim import ValidationFailedError, resilient
  
  
  @resilient(retries=2, retry_on=(ValidationFailedError,), validation=response_validation)
  async def generate(prompt: str) -> str:
      return await provider.generate(prompt)

• it can trigger the fallback, whose candidates are validated with the same policy. An invalid candidate result counts as a candidate failure, so the next candidate is tried;

• it counts as a circuit-breaker failure when it matches record_failure_on;

• if nothing recovers, OperationExecutionError.cause is the ValidationFailedError, or a FallbackChainError when the fallback chain ran and failed.

Validation checks values; it never transforms them. The result you receive is exactly what the operation returned.

Reports and events#

Report metadata and every attempt’s metadata describe the outcome of the last validation that ran:

Key	Meaning
validation_performed	Whether the value was validated in this attempt or execution.
validation_valid	Whether it passed.
validation_validator_name	The failing validator, or the last validator when all passed.
validation_reason	The failure reason, or None.

With structured events, each validation emits validation.succeeded or validation.failed. For fallback candidates the event also carries fallback_candidate_name.

Reasons are stored in reports and events. Do not put secrets in validation messages, and see the privacy note for JSON Schema reasons.

Schema and model validators#

If your responses already have a JSON Schema or a Pydantic model, use an optional adapter. Both implement the same Validator protocol and go into the same policy, alongside callable validators:

from relprim import validation_policy
from relprim.jsonschema import json_schema_validator
from relprim.pydantic import pydantic_validator

response_validation = validation_policy(
    json_schema_validator(RESPONSE_SCHEMA, name="response_schema"),
    pydantic_validator(ProviderResponse),
)

• JSON Schema: pip install "relprim[jsonschema]"
• Pydantic: pip install "relprim[pydantic]"

Writing a validator#

Any object with a read-only name and a validate(value) -> ValidationResult method is a Validator:

from relprim import ValidationResult


class MaxLength:
    def __init__(self, limit: int) -> None:
        self._limit = limit

    @property
    def name(self) -> str:
        return "max_length"

    def validate(self, value: str) -> ValidationResult:
        if len(value) <= self._limit:
            return ValidationResult.success(validator_name=self.name)
        return ValidationResult.failure(
            validator_name=self.name,
            reason=f"Response exceeds {self._limit} characters.",
        )

A failed result must include a reason; a successful one must not. Validators run synchronously inside the attempt, so keep them fast and free of I/O.

Typing#

Validator, CallableValidator and ValidationPolicy are contravariant: a policy that accepts object, such as one built from the JSON Schema adapter, can be used with an operation of any return type. With the builder, .with_validation(...) is checked against the operation’s return type; the decorator checks compatibility at runtime only.