Building Validation Pipelines

Validation pipelines allow you to chain multiple validation and transformation steps into a single execution sequence. In this tutorial, you will build a currency processing pipeline that cleans raw string input, converts it to a Decimal type, and enforces numerical constraints.

You will use the low-level ChainSchema from pydantic-core and the high-level experimental pipeline API from pydantic.

Prerequisites

To follow this tutorial, you need pydantic and pydantic-core installed. The pipeline API is located in the experimental module:

from pydantic.experimental.pipeline import validate_as, transform

Step 1: Building a Low-Level Chain

The foundation of validation pipelines is the ChainSchema. It executes a list of CoreSchema steps in order, passing the output of one step as the input to the next.

Create a basic validator that ensures an input is a string and then appends a suffix using pydantic_core.core_schema.chain_schema:

from pydantic_core import SchemaValidator, core_schema

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

# Define the steps in the chain
schema = core_schema.chain_schema(
    steps=[
        core_schema.str_schema(),  # Step 1: Ensure input is a string
        core_schema.no_info_plain_validator_function(add_suffix),  # Step 2: Transform
    ]
)

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

In this example, chain_schema takes a list of schemas. If the first step (string validation) fails, the chain stops. If it succeeds, the resulting string is passed to the add_suffix function.

Step 2: Using the Fluent Pipeline API

While chain_schema is powerful, it can become verbose for complex sequences. The pydantic.experimental.pipeline module provides a fluent interface that compiles down to a ChainSchema.

Build a pipeline that strips whitespace from a string and then validates it as an integer:

from typing import Annotated
from pydantic import TypeAdapter
from pydantic.experimental.pipeline import validate_as

# Define a pipeline: validate as str -> strip -> validate as int
pipeline = validate_as(str).str_strip().validate_as(int)

# Use TypeAdapter to run the pipeline
ta = TypeAdapter[int](Annotated[int, pipeline])

print(ta.validate_python("  123  "))
# Output: 123

The validate_as(str) call initializes the pipeline. The .str_strip() method adds a transformation step, and the final .validate_as(int) ensures the resulting stripped string can be parsed into an integer.

Step 3: Creating a Complex Currency Processor

Now, let's combine these concepts to build a robust currency processor. This pipeline will:

1. Accept a string.
2. Strip whitespace.
3. Convert the string to a Decimal.
4. Ensure the value is greater than zero.

from decimal import Decimal
from typing import Annotated
from pydantic import BaseModel, Field, ValidationError
from pydantic.experimental.pipeline import validate_as

# Define the reusable currency pipeline
currency_pipeline = (
    validate_as(str)
    .str_strip()
    .validate_as(Decimal)
    .gt(Decimal("0"))
)

class Product(BaseModel):
    name: str
    # Apply the pipeline using Annotated
    price: Annotated[Decimal, currency_pipeline]

# Test with valid input
p = Product(name="Laptop", price="  999.99  ")
print(f"{p.name} costs {p.price} ({type(p.price)})")
# Output: Laptop costs 999.99 (<class 'decimal.Decimal'>)

# Test with invalid input (negative value)
try:
    Product(name="Cheap Item", price="-10.00")
except ValidationError as e:
    print(e)
    # Output: Input should be greater than 0

How it Works

1. validate_as(str): Ensures the initial input is a string.
2. .str_strip(): Internally adds a no_info_plain_validator_function to the ChainSchema that calls .strip() on the input.
3. .validate_as(Decimal): Adds a step that uses Pydantic's internal Decimal validator. Because the previous step output a string, this validator parses that string into a Decimal object.
4. .gt(Decimal("0")): Adds a constraint step (using annotated_types.Gt) that is checked against the Decimal object.

Important Considerations

• Step Order: The order of steps in ChainSchema is critical. If you placed .gt() before .validate_as(Decimal), the validation would fail because you cannot perform a "greater than" comparison on a raw string against a Decimal.
• Empty Chains: A ChainSchema must have at least one step. Providing an empty list to chain_schema will result in a SchemaError.
• Experimental Status: The pydantic.experimental.pipeline API is subject to change. For production-critical code where stability is paramount, consider using the lower-level core_schema.chain_schema or standard AfterValidator and BeforeValidator patterns.

Next Steps

To further customize your pipelines, you can use the transform() function to inject arbitrary Python logic between validation steps, or use the | operator to create unions (branching logic) within your pipeline.