valid8r.async_validators

Async validator library for I/O-bound validation operations.

This module provides validators for operations that require async I/O, such as database queries, API calls, and DNS lookups.

All validators return Maybe[T] and follow the same composable pattern as synchronous validators. This allows for efficient validation against external systems without blocking the event loop.

Key Features:

• Database validation (unique_in_db, exists_in_db)

• API validation (valid_api_key, valid_oauth_token)

• Email deliverability (valid_email_deliverable)

• Rate limiting (RateLimitedValidator)

• Batch validation (parallel_validate)

• Non-blocking async operations

• Compatible with Maybe monad pattern

• Works with any async database connection

Example

>>> import asyncio
>>> from valid8r.async_validators import unique_in_db
>>>
>>> async def example():
...     # Create validator for checking email uniqueness
...     validator = await unique_in_db(
...         field='email',
...         table='users',
...         connection=db_conn
...     )
...
...     # Validate that email is unique
...     result = await validator('new@example.com')
...     if result.is_success():
...         print(f"Email {result.value_or(None)} is available!")
...     else:
...         print(f"Error: {result.error_or('')}")

Attributes

AsyncValidator

Classes

AsyncCache	Protocol for async cache implementations.
DNSResolver	Protocol for DNS resolver implementations.
RateLimitedValidator	A wrapper that adds rate limiting to async validators.
RetryValidator	A wrapper that adds retry logic to async validators.
RetryingValidator	Wrap async validator with retry logic and exponential backoff.
TimeoutValidator	Wrap async validator with timeout handling.
CachedValidator	Wrap async validator with TTL-based result caching.

Functions

unique_in_db(*, field, table, connection)	Create a validator that checks if a value is unique in a database table.
exists_in_db(*, field, table, connection)	Create a validator that checks if a value exists in a database table.
valid_api_key(*, api_url[, timeout, verifier, cache])	Create a validator that checks if an API key is valid against an external service.
valid_oauth_token(*, token_endpoint[, cache, verifier])	Create a validator that checks if an OAuth token is valid.
valid_email_deliverable(*[, resolver])	Create a validator that checks if an email address is deliverable.
parallel_validate(validator, values[, max_concurrency])	Validate multiple values concurrently.
sequential_validate(validators, value)	Run validators sequentially, stopping on first failure.
compose_parallel(validators, value)	Run validators in parallel and combine results.
all_of(*validators[, fail_fast])	Create a composed validator that requires all validators to pass.
any_of(*validators)	Create a composed validator where at least one validator must pass.
sequence(*validators)	Create a composed validator that runs validators sequentially.

Module Contents

valid8r.async_validators.AsyncValidator

class valid8r.async_validators.AsyncCache

Bases: Protocol

Protocol for async cache implementations.

async get(key)

Get value from cache.

Parameters:

key (str)

Return type:

Any | None

async set(key, value)

Set value in cache.

Parameters:

• key (str)

• value (Any)

Return type:

None

class valid8r.async_validators.DNSResolver

Bases: Protocol

Protocol for DNS resolver implementations.

async resolve_mx(domain)

Resolve MX records for domain.

Parameters:

domain (str)

Return type:

list[str]

async valid8r.async_validators.unique_in_db(*, field, table, connection)

Create a validator that checks if a value is unique in a database table.

This validator queries the database to ensure the value doesn’t already exist in the specified field of the specified table. Use this when validating user input that must be unique, such as email addresses, usernames, or identifiers.

The validator executes a COUNT query against the database and returns a Failure if the value already exists, or Success if it’s unique.

Parameters:

• field (str) – The database field/column to check (e.g., ‘email’, ‘username’)

• table (str) – The database table to query (e.g., ‘users’, ‘accounts’)

• connection (Any) – An async database connection object with an execute() method that returns a result with a scalar() method. Compatible with asyncpg, aiopg, and similar async database libraries.

Returns:

• Accepts a value to validate

• Returns Maybe[Any]: Success(value) if unique, Failure(error_msg) if not

• Returns Failure for database errors

Return type:

An async validator function that

Example

>>> import asyncio
>>> import asyncpg
>>> from valid8r.async_validators import unique_in_db
>>>
>>> async def validate_new_user_email():
...     # Connect to database
...     conn = await asyncpg.connect('postgresql://localhost/mydb')
...
...     # Create validator
...     email_validator = await unique_in_db(
...         field='email',
...         table='users',
...         connection=conn
...     )
...
...     # Validate email uniqueness
...     result = await email_validator('new@example.com')
...     if result.is_success():
...         print(f"Email is available: {result.value_or(None)}")
...     else:
...         print(f"Email taken: {result.error_or('')}")
...
...     await conn.close()
>>>
>>> asyncio.run(validate_new_user_email())

Notes

• The validator is non-blocking and safe to use in async frameworks

• Database errors are caught and returned as Failure results

• The field and table names are interpolated into the SQL query

• Use parameterized queries to prevent SQL injection

async valid8r.async_validators.exists_in_db(*, field, table, connection)

Create a validator that checks if a value exists in a database table.

This validator queries the database to ensure the value exists in the specified field of the specified table. Use this when validating foreign keys, references, or ensuring that a related entity exists before proceeding.

The validator executes a COUNT query against the database and returns a Failure if the value doesn’t exist, or Success if it does.

Parameters:

• field (str) – The database field/column to check (e.g., ‘id’, ‘category_id’)

• table (str) – The database table to query (e.g., ‘categories’, ‘users’)

• connection (Any) – An async database connection object with an execute() method that returns a result with a scalar() method. Compatible with asyncpg, aiopg, and similar async database libraries.

Returns:

• Accepts a value to validate

• Returns Maybe[Any]: Success(value) if exists, Failure(error_msg) if not

• Returns Failure for database errors

Return type:

An async validator function that

Example

>>> import asyncio
>>> import asyncpg
>>> from valid8r.async_validators import exists_in_db
>>>
>>> async def validate_category_reference():
...     # Connect to database
...     conn = await asyncpg.connect('postgresql://localhost/mydb')
...
...     # Create validator for category_id foreign key
...     category_validator = await exists_in_db(
...         field='id',
...         table='categories',
...         connection=conn
...     )
...
...     # Validate that category exists
...     result = await category_validator('electronics')
...     if result.is_success():
...         print(f"Category exists: {result.value_or(None)}")
...     else:
...         print(f"Invalid category: {result.error_or('')}")
...
...     await conn.close()
>>>
>>> asyncio.run(validate_category_reference())

Notes

• The validator is non-blocking and safe to use in async frameworks

• Database errors are caught and returned as Failure results

• The field and table names are interpolated into the SQL query

• Use parameterized queries to prevent SQL injection

async valid8r.async_validators.valid_api_key(*, api_url, timeout=None, verifier=None, cache=None)

Create a validator that checks if an API key is valid against an external service.

This validator calls an external API endpoint to validate API keys. Use this when validating API keys before processing requests that require authentication.

Parameters:

• api_url (str) – The URL of the API endpoint to validate keys against

• timeout (float | None) – Optional timeout in seconds for the API call

• verifier (APIVerifier | None) – Optional custom API verifier (for testing/mocking)

• cache (AsyncCache | None) – Optional async cache for storing validation results

Returns:

• Accepts an API key to validate

• Returns Maybe[str]: Success(key) if valid, Failure(error_msg) if not

• Returns Failure for network errors or timeouts

Return type:

An async validator function that

Example

>>> import asyncio
>>> from valid8r.async_validators import valid_api_key
>>>
>>> async def validate_key():
...     validator = await valid_api_key(
...         api_url='https://api.example.com/validate',
...         timeout=5.0
...     )
...     result = await validator('my-api-key-123')
...     if result.is_success():
...         print("API key is valid!")
...     else:
...         print(f"Invalid: {result.error_or('')}")

async valid8r.async_validators.valid_oauth_token(*, token_endpoint, cache=None, verifier=None)

Create a validator that checks if an OAuth token is valid.

This validator calls an OAuth token endpoint to validate tokens. Optionally supports caching to avoid redundant API calls for the same token.

Parameters:

• token_endpoint (str) – The URL of the OAuth token validation endpoint

• cache (AsyncCache | None) – Optional async cache for storing validation results

• verifier (APIVerifier | None) – Optional custom API verifier (for testing/mocking)

Returns:

• Accepts an OAuth token to validate

• Returns Maybe[str]: Success(token) if valid, Failure(error_msg) if not

• Uses cache if provided to avoid redundant calls

Return type:

An async validator function that

Example

>>> import asyncio
>>> from valid8r.async_validators import valid_oauth_token
>>>
>>> async def validate_token():
...     validator = await valid_oauth_token(
...         token_endpoint='https://oauth.example.com/token'
...     )
...     result = await validator('bearer-token-123')
...     if result.is_success():
...         print("Token is valid!")

async valid8r.async_validators.valid_email_deliverable(*, resolver=None)

Create a validator that checks if an email address is deliverable.

This validator checks if the email domain has valid MX records, indicating that the domain can receive email. Use this for validating email addresses beyond just format checking.

Parameters:

resolver (DNSResolver | None) – Optional DNS resolver (for testing/mocking)

Returns:

• Accepts an email address (string or EmailAddress object)

• Returns Maybe[EmailAddress]: Success if deliverable, Failure if not

Return type:

An async validator function that

Example

>>> import asyncio
>>> from valid8r.async_validators import valid_email_deliverable
>>>
>>> async def check_email():
...     validator = await valid_email_deliverable()
...     result = await validator('user@example.com')
...     if result.is_success():
...         print("Email domain can receive mail!")

class valid8r.async_validators.RateLimitedValidator(validator, *, rate, burst=None)

Bases: Generic[T]

A wrapper that adds rate limiting to async validators.

Uses a token bucket algorithm to limit the rate of validation calls. This is useful for protecting external services from being overwhelmed.

Parameters:

• validator (AsyncValidator) – The async validator function to wrap

• rate (int) – Maximum number of calls per second

• burst (int | None) – Maximum burst size (defaults to rate)

Example

>>> import asyncio
>>> from valid8r.async_validators import RateLimitedValidator
>>>
>>> async def my_validator(value):
...     return Maybe.success(value)
>>>
>>> rate_limited = RateLimitedValidator(my_validator, rate=10, burst=5)
>>> result = await rate_limited('test')

async __call__(value)

Validate a value with rate limiting.

Parameters:

value (T) – The value to validate

Returns:

The validation result

Return type:

Maybe[T]

class valid8r.async_validators.RetryValidator(validator, *, max_retries=3, base_delay=0.1, exponential=True)

Bases: Generic[T]

A wrapper that adds retry logic to async validators.

Retries validation on transient failures with configurable backoff.

Parameters:

• validator (AsyncValidator) – The async validator function to wrap

• max_retries (int) – Maximum number of retry attempts (default: 3)

• base_delay (float) – Base delay in seconds for exponential backoff (default: 0.1)

• exponential (bool) – Use exponential backoff if True (default: True)

Example

>>> import asyncio
>>> from valid8r.async_validators import RetryValidator
>>>
>>> retry_validator = RetryValidator(my_validator, max_retries=3)
>>> result = await retry_validator('test')

retry_count = 0

retry_delays: list[float] = []

async __call__(value)

Validate a value with retry logic.

Parameters:

value (T) – The value to validate

Returns:

The validation result

Return type:

Maybe[T]

class valid8r.async_validators.RetryingValidator(validator, max_retries=3, base_delay=1.0, max_delay=60.0, exponential_base=2.0, *, jitter=True)

Bases: Generic[T]

Wrap async validator with retry logic and exponential backoff.

This validator wrapper adds robust retry handling for transient failures with configurable exponential backoff and optional jitter. Use this to make async validators resilient to temporary network issues, rate limits, or service unavailability.

The retry logic: 1. Attempts the validation 2. On failure, waits with exponential backoff before retrying 3. Optionally adds jitter to prevent thundering herd problems 4. Caps delay at max_delay to prevent excessive waits 5. After max_retries exhausted, returns failure with last error

Parameters:

• validator (collections.abc.Callable[[T], collections.abc.Awaitable[valid8r.core.maybe.Maybe[T]]]) – The async validator function to wrap. Must be a callable that takes a value and returns Awaitable[Maybe[T]].

• max_retries (int) – Maximum number of retry attempts after the initial call. Default: 3 (4 total attempts including initial).

• base_delay (float) – Base delay in seconds for exponential backoff. The actual delay is base_delay * exponential_base^attempt. Default: 1.0.

• max_delay (float) – Maximum delay in seconds. Delays are capped at this value to prevent excessive waiting. Default: 60.0.

• exponential_base (float) – Base for exponential backoff calculation. A value of 2.0 doubles the delay each retry. Default: 2.0.

• jitter (bool) – If True, adds random jitter to delays to prevent thundering herd when multiple validators retry simultaneously. Default: True.

retry_count

Number of retries performed in the last validation call.

retry_delays

List of actual delays (in seconds) used between retries.

Example

>>> import asyncio
>>> from valid8r.async_validators import RetryingValidator
>>> from valid8r.core.maybe import Maybe
>>>
>>> async def flaky_api_validator(value: str) -> Maybe[str]:
...     # Simulates a validator that might fail transiently
...     import random
...     if random.random() < 0.5:
...         return Maybe.failure('Transient: service unavailable')
...     return Maybe.success(value)
>>>
>>> async def main():
...     # Wrap with retry logic
...     robust_validator = RetryingValidator(
...         flaky_api_validator,
...         max_retries=3,
...         base_delay=0.5,
...         max_delay=5.0,
...         jitter=True
...     )
...     result = await robust_validator('test-value')
...     print(f'Success: {result.is_success()}')
>>>
>>> asyncio.run(main())

Notes

• All failures are retried, not just those containing “transient”

• Exceptions during validation are caught and converted to retries

• The retry_count and retry_delays attributes are reset on each call

retry_count = 0

retry_delays: list[float] = []

async __call__(value)

Validate a value with retry logic.

Attempts validation up to max_retries times on failure, using exponential backoff between attempts.

Parameters:

value (T) – The value to validate

Returns:

Success with validated value, or Failure with error

message including “max retries exceeded” if all attempts fail

Return type:

Maybe[T]

class valid8r.async_validators.TimeoutValidator(validator, timeout)

Bases: Generic[T]

Wrap async validator with timeout handling.

This validator wrapper adds a timeout to any async validator, ensuring that slow validations fail gracefully instead of hanging indefinitely. Use this when calling external services that may become unresponsive.

The wrapper uses asyncio.wait_for() internally and returns a Failure with a descriptive error message when the timeout is exceeded.

Parameters:

• validator (collections.abc.Callable[[T], collections.abc.Awaitable[valid8r.core.maybe.Maybe[T]]]) – The async validator function to wrap. Must be a callable that takes a value and returns Awaitable[Maybe[T]].

• timeout (float) – Maximum time in seconds to wait for the validator to complete. If the validator takes longer than this, a Failure is returned.

Example

>>> import asyncio
>>> from valid8r.async_validators import TimeoutValidator
>>> from valid8r.core.maybe import Maybe
>>>
>>> async def slow_api_validator(value: str) -> Maybe[str]:
...     await asyncio.sleep(10)  # Simulates slow API call
...     return Maybe.success(value)
>>>
>>> async def main():
...     # Wrap with 2 second timeout
...     validator = TimeoutValidator(slow_api_validator, timeout=2.0)
...     result = await validator('test-value')
...     if result.is_failure():
...         print(result.error_or(''))  # "Validation timeout after 2.0s"
>>>
>>> asyncio.run(main())

Notes

• The original validator’s Failure results pass through unchanged

• Timeout error messages include the configured timeout duration

• Can be composed with other validator wrappers (RateLimited, Cached, etc.)

async __call__(value)

Validate a value with timeout.

Attempts to run the wrapped validator within the configured timeout. If the validator completes in time, its result is returned unchanged. If it times out, a Failure with a descriptive message is returned.

Parameters:

value (T) – The value to validate

Returns:

Success or Failure from the validator, or Failure on timeout

Return type:

Maybe[T]

class valid8r.async_validators.CachedValidator(validator, ttl=300.0, key_func=None)

Bases: Generic[T]

Wrap async validator with TTL-based result caching.

This validator wrapper caches successful validation results to avoid redundant calls to external services. Use this when validating values that don’t change frequently and where the validation is expensive.

The cache uses time-based expiration (TTL) and only caches successful validations. Failed validations are not cached to allow retries.

Parameters:

• validator (collections.abc.Callable[[T], collections.abc.Awaitable[valid8r.core.maybe.Maybe[T]]]) – The async validator function to wrap. Must be a callable that takes a value and returns Awaitable[Maybe[T]].

• ttl (float) – Time-to-live in seconds for cached results. After this time, the cached result expires and the validator is called again. Default: 300.0 (5 minutes).

• key_func (collections.abc.Callable[[T], str] | None) – Optional function to generate cache keys from values. By default, uses str(value). Custom key functions are useful when validating complex objects where only certain fields matter.

call_count

Number of times the wrapped validator has been called (excludes cache hits). Useful for testing and monitoring.

Return type:

int

Example

>>> import asyncio
>>> from valid8r.async_validators import CachedValidator
>>> from valid8r.core.maybe import Maybe
>>>
>>> async def expensive_api_validator(value: str) -> Maybe[str]:
...     print(f'API call for {value}')
...     return Maybe.success(value)
>>>
>>> async def main():
...     validator = CachedValidator(expensive_api_validator, ttl=60.0)
...
...     # First call - hits the API
...     await validator('test')  # Prints: "API call for test"
...
...     # Second call - uses cache
...     await validator('test')  # No print, cached
...
...     print(f'API calls: {validator.call_count}')  # "API calls: 1"
>>>
>>> asyncio.run(main())

Notes

• Only successful validations are cached

• Failed validations are NOT cached (allows immediate retry)

• Cache is in-memory and not shared between instances

• Use invalidate() to manually remove a cached entry

• Use clear() to remove all cached entries

property call_count: int

Number of times the wrapped validator was called (excluding cache hits).

Return type:

int

async __call__(value)

Validate a value with caching.

Checks the cache first. On cache miss or expiration, calls the wrapped validator and caches successful results.

Parameters:

value (T) – The value to validate

Returns:

Cached or fresh validation result

Return type:

Maybe[T]

invalidate(value)

Manually invalidate a cached result.

Removes the cached entry for the given value, if it exists. The next validation will call the wrapped validator.

Parameters:

value (T) – The value whose cached result should be invalidated

Return type:

None

clear()

Clear all cached results.

Removes all entries from the cache. All subsequent validations will call the wrapped validator.

Return type:

None

async valid8r.async_validators.parallel_validate(validator, values, max_concurrency=None)

Validate multiple values concurrently.

This helper function validates a sequence of values in parallel, returning all results (both successes and failures). Results are returned in the same order as the input values.

Optionally limits the number of concurrent validations using a semaphore, which is useful for protecting external services from being overwhelmed.

Parameters:

• validator (AsyncValidator) – The async validator function to use

• values (collections.abc.Sequence[T]) – Sequence of values to validate

• max_concurrency (int | None) – Optional maximum number of concurrent validations. If None, all validations run in parallel. If specified, limits concurrent validations to this number.

Returns:

List of Maybe[T] results in the same order as input values

Return type:

list[valid8r.core.maybe.Maybe[T]]

Example

>>> import asyncio
>>> from valid8r.async_validators import parallel_validate
>>>
>>> async def validate_emails():
...     emails = ['a@example.com', 'b@example.com', 'c@example.com']
...
...     # All in parallel
...     results = await parallel_validate(email_validator, emails)
...
...     # Limited to 2 concurrent validations
...     results = await parallel_validate(
...         email_validator, emails, max_concurrency=2
...     )
...
...     successes = [r for r in results if r.is_success()]
...     failures = [r for r in results if r.is_failure()]

Notes

• Results maintain the same order as input values

• All validations complete before returning (uses gather)

• max_concurrency=None means unlimited parallelism

• Use max_concurrency to protect rate-limited external services

async valid8r.async_validators.sequential_validate(validators, value)

Run validators sequentially, stopping on first failure.

This helper function runs a sequence of validators one after another, passing the value through each validator. Stops on first failure.

Parameters:

• validators (collections.abc.Sequence[AsyncValidator]) – Sequence of async validators to run

• value (T) – The value to validate

Returns:

Success if all validators pass, first Failure otherwise

Return type:

Maybe[T]

Example

>>> import asyncio
>>> from valid8r.async_validators import sequential_validate
>>>
>>> async def validate_user():
...     validators = [format_validator, uniqueness_validator, auth_validator]
...     result = await sequential_validate(validators, user_data)

async valid8r.async_validators.compose_parallel(validators, value)

Run validators in parallel and combine results.

All validators run concurrently. Returns Success if all pass, or the first Failure encountered.

Parameters:

• validators (collections.abc.Sequence[AsyncValidator]) – Sequence of async validators to run in parallel

• value (T) – The value to validate

Returns:

Success if all validators pass, first Failure otherwise

Return type:

Maybe[T]

valid8r.async_validators.all_of(*validators, fail_fast=True)

Create a composed validator that requires all validators to pass.

Runs all validators in parallel for efficiency. All validators must succeed for the overall validation to succeed. This is useful when you have multiple independent validation rules that must all be satisfied.

Parameters:

• *validators (AsyncValidator) – Variable number of async validators to compose

• fail_fast (bool) – If True (default), returns the first error encountered. If False, collects all errors and returns them joined together.

Returns:

• Accepts a value to validate

• Returns Maybe[T]: Success(value) if all validators pass, Failure(error) if any validator fails

Return type:

A callable async validator that

Example

>>> import asyncio
>>> from valid8r.async_validators import all_of
>>> from valid8r.core.maybe import Maybe
>>>
>>> async def check_length(value: str) -> Maybe[str]:
...     if len(value) >= 3:
...         return Maybe.success(value)
...     return Maybe.failure('Too short')
>>>
>>> async def check_alpha(value: str) -> Maybe[str]:
...     if value.isalpha():
...         return Maybe.success(value)
...     return Maybe.failure('Not alphabetic')
>>>
>>> async def main():
...     validator = all_of(check_length, check_alpha)
...     result = await validator('hello')
...     print(f'Success: {result.is_success()}')
>>>
>>> asyncio.run(main())

Notes

• Validators run in parallel using asyncio.gather

• Empty validator list returns Success with the original value

• When fail_fast=False, errors are joined with ‘; ‘

valid8r.async_validators.any_of(*validators)

Create a composed validator where at least one validator must pass.

Runs all validators in parallel for efficiency. At least one validator must succeed for the overall validation to succeed. This is useful for alternative validation paths or fallback validation logic.

Parameters:

*validators (AsyncValidator) – Variable number of async validators to compose

Returns:

• Accepts a value to validate

• Returns Maybe[T]: Success(value) if any validator passes, Failure(error) if all validators fail

Return type:

A callable async validator that

Example

>>> import asyncio
>>> from valid8r.async_validators import any_of
>>> from valid8r.core.maybe import Maybe
>>>
>>> async def check_email(value: str) -> Maybe[str]:
...     if '@' in value:
...         return Maybe.success(value)
...     return Maybe.failure('Not an email')
>>>
>>> async def check_username(value: str) -> Maybe[str]:
...     if value.isalnum():
...         return Maybe.success(value)
...     return Maybe.failure('Not a valid username')
>>>
>>> async def main():
...     validator = any_of(check_email, check_username)
...     # Either email or username format is acceptable
...     result = await validator('user123')
...     print(f'Success: {result.is_success()}')
>>>
>>> asyncio.run(main())

Notes

• Validators run in parallel using asyncio.gather

• Empty validator list returns Failure (nothing can succeed)

• Returns the result from the first successful validator

• If all fail, errors are joined with ‘; ‘

valid8r.async_validators.sequence(*validators)

Create a composed validator that runs validators sequentially.

Runs validators one after another, passing the result of each validator to the next. Stops on first failure. This is useful when validators have dependencies or when the output of one validator feeds into the next.

Parameters:

*validators (AsyncValidator) – Variable number of async validators to compose

Returns:

• Accepts a value to validate

• Returns Maybe[T]: Success(final_value) if all validators pass, Failure(error) on first failure

Return type:

A callable async validator that

Example

>>> import asyncio
>>> from valid8r.async_validators import sequence
>>> from valid8r.core.maybe import Maybe
>>>
>>> async def trim_whitespace(value: str) -> Maybe[str]:
...     return Maybe.success(value.strip())
>>>
>>> async def to_lowercase(value: str) -> Maybe[str]:
...     return Maybe.success(value.lower())
>>>
>>> async def validate_length(value: str) -> Maybe[str]:
...     if len(value) >= 3:
...         return Maybe.success(value)
...     return Maybe.failure('Too short')
>>>
>>> async def main():
...     validator = sequence(trim_whitespace, to_lowercase, validate_length)
...     result = await validator('  HELLO  ')
...     print(f'Result: {result.value_or("")}')  # 'hello'
>>>
>>> asyncio.run(main())

Notes

• Validators run sequentially, NOT in parallel

• Each validator receives the SUCCESS value from the previous one

• Stops execution on first Failure

• Empty validator list returns Success with the original value