valid8r.core.schema

Schema-based validation with error accumulation and field path tracking.

This module provides the Schema and Field classes for validating dict-like objects against a defined schema, accumulating all validation errors instead of stopping at the first failure.

Key features: - Error accumulation across multiple fields - Field path tracking (e.g., “.age”, “.user.email”) - Nested schema composition - Required/optional field support - Strict mode for rejecting extra fields - Integration with existing parsers and validators

Examples

Basic schema validation:

>>> from valid8r.core import parsers, schema
>>> from valid8r.core.maybe import Success, Failure
>>> s = schema.Schema(fields={
...     'age': schema.Field(parser=parsers.parse_int, required=True),
...     'email': schema.Field(parser=parsers.parse_email, required=True),
... })
>>> result = s.validate({'age': '25', 'email': 'alice@example.com'})
>>> match result:
...     case Success(data):
...         print(f"Age: {data['age']}, Email: {data['email']}")
...     case Failure(errors):
...         print(f"Errors: {errors}")
Age: 25, Email: EmailAddress(local='alice', domain='example.com')

Error accumulation:

>>> result = s.validate({'age': 'invalid', 'email': 'bad'})
>>> result.is_failure()
True

Classes

Field	Schema field definition with parser, validators, and required flag.
Schema	Schema for validating dict-like objects with error accumulation.

Module Contents

class valid8r.core.schema.Field

Schema field definition with parser, validators, and required flag.

A Field represents a single field in a schema, specifying how to parse and validate the field value.

parser

Function that parses/validates the raw value, returns Maybe[T]

validators

Optional list of validation functions to apply after parsing

required

Whether the field must be present in the input

Examples

Required field with just a parser:

>>> from valid8r.core import parsers
>>> field = Field(parser=parsers.parse_int, required=True)
>>> field.required
True

Optional field:

>>> field = Field(parser=parsers.parse_str, required=False)
>>> field.required
False

Field with parser and validators:

>>> from valid8r.core import validators
>>> field = Field(
...     parser=parsers.parse_int,
...     validators=[validators.minimum(0), validators.maximum(100)],
...     required=True
... )
>>> len(field.validators)
2

parser: collections.abc.Callable[[Any], valid8r.core.maybe.Maybe[Any]]

required: bool

validators: list[collections.abc.Callable[[Any], valid8r.core.maybe.Maybe[Any]]] | None = None

class valid8r.core.schema.Schema(*, fields, strict=False)

Schema for validating dict-like objects with error accumulation.

A Schema defines the structure and validation rules for a dict-like object, accumulating all validation errors across all fields instead of stopping at the first failure.

Parameters:

• fields (dict[str, Field])

• strict (bool)

fields

Dictionary mapping field names to Field definitions

strict

If True, reject inputs with fields not defined in the schema

Examples

Basic schema validation:

>>> from valid8r.core import parsers
>>> s = Schema(fields={
...     'age': Field(parser=parsers.parse_int, required=True),
...     'name': Field(parser=parsers.parse_str, required=True),
... })
>>> result = s.validate({'age': '25', 'name': 'Alice'})
>>> result.is_success()
True

Error accumulation:

>>> from valid8r.core.maybe import Failure
>>> result = s.validate({'age': 'bad', 'name': ''})
>>> result.is_failure()
True

Strict mode:

>>> strict_schema = Schema(
...     fields={'name': Field(parser=parsers.parse_str, required=True)},
...     strict=True
... )
>>> result = strict_schema.validate({'name': 'Alice', 'extra': 'field'})
>>> result.is_failure()
True

fields

strict = False

validate(data, path='')

Validate data against the schema, accumulating all errors.

This method validates the input data against all field definitions, collecting all validation errors instead of stopping at the first failure. Field paths are tracked for nested validation (e.g., “.user.email”).

Parameters:

• data (dict[str, Any] | Any) – Input data to validate (must be dict-like)

• path (str) – Current field path for nested validation (internal use)

Returns:

Validated and parsed data if all fields pass Failure[list[ValidationError]]: List of all validation errors

Return type:

Success[dict]

Examples

Successful validation:

>>> from valid8r.core import parsers
>>> from valid8r.core.maybe import Success
>>> s = Schema(fields={
...     'age': Field(parser=parsers.parse_int, required=True),
... })
>>> result = s.validate({'age': '30'})
>>> match result:
...     case Success(data):
...         data['age']
...     case _:
...         None
30

Multiple errors:

>>> s = Schema(fields={
...     'age': Field(parser=parsers.parse_int, required=True),
...     'email': Field(parser=parsers.parse_email, required=True),
... })
>>> result = s.validate({'age': 'bad', 'email': 'bad'})
>>> result.is_failure()
True

async validate_async(data, path='', *, timeout=None)

Validate data against the schema asynchronously with async validators.

This method validates input data supporting both sync and async validators. Sync validators are run first for fail-fast behavior, then async validators are run concurrently for better performance.

Parameters:

• data (dict[str, Any] | Any) – Input data to validate (must be dict-like)

• path (str) – Current field path for nested validation (internal use)

• timeout (float | None) – Optional timeout in seconds for async operations

Returns:

Validated and parsed data if all fields pass Failure[list[ValidationError]]: List of all validation errors

Return type:

Success[dict]

Raises:

asyncio.TimeoutError – If validation exceeds the timeout

Examples

Basic async validation:

>>> import asyncio
>>> from valid8r.core import parsers, schema
>>> from valid8r.core.maybe import Maybe
>>> async def async_validator(val: str) -> Maybe[str]:
...     await asyncio.sleep(0.001)
...     return Maybe.success(val)
>>> s = schema.Schema(fields={
...     'field': schema.Field(
...         parser=parsers.parse_str,
...         validators=[async_validator],
...         required=True
...     ),
... })
>>> result = asyncio.run(s.validate_async({'field': 'value'}))
>>> result.is_success()
True

With timeout:

>>> async def slow_validator(val: str) -> Maybe[str]:
...     await asyncio.sleep(2.0)
...     return Maybe.success(val)
>>> s = schema.Schema(fields={
...     'field': schema.Field(
...         parser=parsers.parse_str,
...         validators=[slow_validator],
...         required=True
...     ),
... })
>>> try:
...     result = asyncio.run(s.validate_async({'field': 'value'}, timeout=0.1))
... except asyncio.TimeoutError:
...     print("Timed out")
Timed out