Implementing Custom Validators

In this tutorial, you will build a robust validation pipeline by extending the validation engine with custom functions. You will learn how to use the four primary validation modes—Before, After, Plain, and Wrap—and how to leverage validation metadata through the ValidationInfo object.

Prerequisites

To follow this tutorial, you need pydantic-core installed. You will be working primarily with the core_schema module and the SchemaValidator class.

from typing import Any
from pydantic_core import core_schema, SchemaValidator

Step 1: Pre-processing Data with Before Validators

A "before" validator runs before the inner schema validation. This is ideal for coercing types or cleaning raw input data before it reaches the standard Pydantic validation logic.

You will create a validator that ensures a string is stripped of whitespace and converted to lowercase before being validated as a standard string.

def clean_string(v: Any) -> str:
    if isinstance(v, str):
        return v.strip().lower()
    return v

# This creates a BeforeValidatorFunctionSchema
schema = core_schema.no_info_before_validator_function(
    function=clean_string,
    schema=core_schema.str_schema()
)

validator = SchemaValidator(schema)
result = validator.validate_python("  EXAMPLE  ")
print(result)
# Output: "example"

In this step, core_schema.no_info_before_validator_function takes your custom function and an inner schema (str_schema). The clean_string function processes the raw input, and its output is then passed to the string schema for final validation.

Step 2: Refining Results with After Validators

An "after" validator runs after the inner schema has successfully validated the data. This is useful for performing complex checks or transformations on already-validated types.

You will implement a validator that appends a suffix to a string after it has passed the initial string validation.

def add_suffix(v: str) -> str:
    return f"{v}_processed"

# This creates an AfterValidatorFunctionSchema
schema = core_schema.no_info_after_validator_function(
    function=add_suffix,
    schema=core_schema.str_schema()
)

validator = SchemaValidator(schema)
result = validator.validate_python("data")
print(result)
# Output: "data_processed"

Here, core_schema.no_info_after_validator_function ensures that add_suffix only receives a value that has already been confirmed as a string by core_schema.str_schema().

Step 3: Intercepting Validation with Wrap Validators

A "wrap" validator gives you complete control over the validation process. It receives a handler (a ValidatorFunctionWrapHandler) which you can call to trigger the inner schema's validation at any point, or even skip it entirely.

You will build a validator that catches validation errors from the inner schema and returns a fallback value instead.

def safe_validate(v: Any, handler: core_schema.ValidatorFunctionWrapHandler) -> int:
    try:
        # Call the inner validator
        return handler(v)
    except Exception:
        # Return a fallback if inner validation fails
        return 0

# This creates a WrapValidatorFunctionSchema
schema = core_schema.no_info_wrap_validator_function(
    function=safe_validate,
    schema=core_schema.int_schema()
)

validator = SchemaValidator(schema)

print(validator.validate_python(10))    # Output: 10
print(validator.validate_python("abc")) # Output: 0

The handler is an instance of ValidatorFunctionWrapHandler. By wrapping the handler(v) call in a try-except block, you can gracefully handle invalid inputs that would otherwise raise a ValidationError.

Step 4: Total Control with Plain Validators

A "plain" validator completely replaces the standard validation logic for a field. It does not have an inner schema.

def custom_logic(v: Any) -> str:
    if v == "secret_code":
        return "access_granted"
    return "access_denied"

# This creates a PlainValidatorFunctionSchema
schema = core_schema.no_info_plain_validator_function(function=custom_logic)

validator = SchemaValidator(schema)
print(validator.validate_python("secret_code")) # Output: "access_granted"

PlainValidatorFunctionSchema is used when you want to bypass Pydantic's built-in types entirely and define your own logic from scratch.

Step 5: Accessing Context with ValidationInfo

Often, validation depends on external state or other fields in a model. The ValidationInfo object provides access to context, data, and other metadata.

You will create a validator that checks a value against a threshold provided in the validation context.

def check_threshold(v: int, info: core_schema.ValidationInfo) -> int:
    # Access context passed during validation
    threshold = info.context.get("min_value", 0) if info.context else 0
    if v < threshold:
        raise ValueError(f"Value {v} is below threshold {threshold}")
    return v

# Use with_info_* to receive the ValidationInfo argument
schema = core_schema.with_info_after_validator_function(
    function=check_threshold,
    schema=core_schema.int_schema()
)

validator = SchemaValidator(schema)

# Pass context to the validate_python method
result = validator.validate_python(10, context={"min_value": 5})
print(result) # Output: 10

# This will raise a ValueError
# validator.validate_python(3, context={"min_value": 5})

By using core_schema.with_info_after_validator_function, your function receives a ValidationInfo instance. This allows you to access info.context (passed at runtime) or info.data (containing other fields if validating a model).

Important Considerations

• ValidationInfo.data: When validating a model, info.data contains the dictionary of fields that have already been validated. Note that it only contains fields defined before the current field in the model.
• Execution Order: When using multiple validators, "Before" and "Wrap" validators generally run from the outside in, while "After" validators run from the inside out.
• Handler Requirement: In a "Wrap" validator, if you do not call the handler, the inner schema validation is skipped entirely.

Complete Working Result

Below is a combined example demonstrating a complex schema using these custom validators to process a user input dictionary.

from typing import Any
from pydantic_core import core_schema, SchemaValidator

def username_before(v: Any) -> str:
    return str(v).strip()

def username_after(v: str) -> str:
    if len(v) < 3:
        raise ValueError("Username too short")
    return v.capitalize()

def password_wrap(v: Any, handler: core_schema.ValidatorFunctionWrapHandler) -> str:
    result = handler(v)
    return "*" * len(result)

schema = core_schema.typed_dict_schema({
    "username": core_schema.typed_dict_field(
        core_schema.no_info_before_validator_function(
            username_before,
            core_schema.no_info_after_validator_function(
                username_after,
                core_schema.str_schema()
            )
        )
    ),
    "password": core_schema.typed_dict_field(
        core_schema.no_info_wrap_validator_function(
            password_wrap,
            core_schema.str_schema()
        )
    )
})

validator = SchemaValidator(schema)
user_data = {"username": "  alice  ", "password": "secret123"}
print(validator.validate_python(user_data))
# Output: {'username': 'Alice', 'password': '*********'}

This pipeline cleans the username, validates its length, capitalizes it, and masks the password—all using the custom validator schemas you've learned.