Experimental
Experimental pipeline API functionality. Be careful with this API, it’s subject to change.
Bases: Generic[_InT, _OutT]
Abstract representation of a chain of validation, transformation, and parsing steps.
def transform(func: Callable[[_OutT], _NewOutT]) -> _Pipeline[_InT, _NewOutT]
Transform the output of the previous step.
If used as the first step in a pipeline, the type of the field is used. That is, the transformation is applied to after the value is parsed to the field’s type.
_Pipeline[_InT, _NewOutT]
def validate_as(tp: type[_NewOutT], strict: bool = False) -> _Pipeline[_InT, _NewOutT]
def validate_as(tp: EllipsisType, strict: bool = False) -> _Pipeline[_InT, Any]
def validate_as(tp: Any, strict: bool = ...) -> _Pipeline[_InT, Any]
Validate / parse the input into a new type.
If no type is provided, the type of the field is used.
Types are parsed in Pydantic’s lax mode by default,
but you can enable strict mode by passing strict=True.
_Pipeline[_InT, Any]
def validate_as_deferred(
func: Callable[[], type[_NewOutT]],
) -> _Pipeline[_InT, _NewOutT]
Parse the input into a new type, deferring resolution of the type until the current class is fully defined.
This is useful when you need to reference the class in it’s own type annotations.
_Pipeline[_InT, _NewOutT]
def constrain(constraint: annotated_types.Ge) -> _Pipeline[_InT, _NewOutGe]
def constrain(constraint: annotated_types.Gt) -> _Pipeline[_InT, _NewOutGt]
def constrain(constraint: annotated_types.Le) -> _Pipeline[_InT, _NewOutLe]
def constrain(constraint: annotated_types.Lt) -> _Pipeline[_InT, _NewOutLt]
def constrain(constraint: annotated_types.Len) -> _Pipeline[_InT, _NewOutLen]
def constrain(constraint: annotated_types.MultipleOf) -> _Pipeline[_InT, _NewOutT]
def constrain(constraint: annotated_types.Timezone) -> _Pipeline[_InT, _NewOutDatetime]
def constrain(constraint: annotated_types.Predicate) -> _Pipeline[_InT, _OutT]
def constrain(constraint: annotated_types.Interval) -> _Pipeline[_InT, _NewOutInterval]
def constrain(constraint: _Eq) -> _Pipeline[_InT, _OutT]
def constrain(constraint: _NotEq) -> _Pipeline[_InT, _OutT]
def constrain(constraint: _In) -> _Pipeline[_InT, _OutT]
def constrain(constraint: _NotIn) -> _Pipeline[_InT, _OutT]
def constrain(constraint: Pattern[str]) -> _Pipeline[_InT, _NewOutT]
Constrain a value to meet a certain condition.
We support most conditions from annotated_types, as well as regular expressions.
Most of the time you’ll be calling a shortcut method like gt, lt, len, etc
so you don’t need to call this directly.
def predicate(func: Callable[[_NewOutT], bool]) -> _Pipeline[_InT, _NewOutT]
Constrain a value to meet a certain predicate.
_Pipeline[_InT, _NewOutT]
def gt(gt: _NewOutGt) -> _Pipeline[_InT, _NewOutGt]
Constrain a value to be greater than a certain value.
_Pipeline[_InT, _NewOutGt]
def lt(lt: _NewOutLt) -> _Pipeline[_InT, _NewOutLt]
Constrain a value to be less than a certain value.
_Pipeline[_InT, _NewOutLt]
def ge(ge: _NewOutGe) -> _Pipeline[_InT, _NewOutGe]
Constrain a value to be greater than or equal to a certain value.
_Pipeline[_InT, _NewOutGe]
def le(le: _NewOutLe) -> _Pipeline[_InT, _NewOutLe]
Constrain a value to be less than or equal to a certain value.
_Pipeline[_InT, _NewOutLe]
def len(min_len: int, max_len: int | None = None) -> _Pipeline[_InT, _NewOutLen]
Constrain a value to have a certain length.
_Pipeline[_InT, _NewOutLen]
def multiple_of(multiple_of: _NewOutDiv) -> _Pipeline[_InT, _NewOutDiv]
def multiple_of(multiple_of: _NewOutMod) -> _Pipeline[_InT, _NewOutMod]
Constrain a value to be a multiple of a certain number.
_Pipeline[_InT, Any]
def eq(value: _OutT) -> _Pipeline[_InT, _OutT]
Constrain a value to be equal to a certain value.
_Pipeline[_InT, _OutT]
def not_eq(value: _OutT) -> _Pipeline[_InT, _OutT]
Constrain a value to not be equal to a certain value.
_Pipeline[_InT, _OutT]
def in_(values: Container[_OutT]) -> _Pipeline[_InT, _OutT]
Constrain a value to be in a certain set.
_Pipeline[_InT, _OutT]
def not_in(values: Container[_OutT]) -> _Pipeline[_InT, _OutT]
Constrain a value to not be in a certain set.
_Pipeline[_InT, _OutT]
def datetime_tz_naive() -> _Pipeline[_InT, datetime.datetime]
Constrain a datetime value to be timezone-naive (i.e. have no tzinfo).
_Pipeline[_InT, datetime.datetime]
def datetime_tz_aware() -> _Pipeline[_InT, datetime.datetime]
Constrain a datetime value to be timezone-aware (i.e. have a non-None tzinfo).
_Pipeline[_InT, datetime.datetime]
def datetime_tz(tz: datetime.tzinfo) -> _Pipeline[_InT, datetime.datetime]
Constrain a datetime value to have a specific timezone.
_Pipeline[_InT, datetime.datetime]
tz : datetime.tzinfo
The required timezone.
def datetime_with_tz(tz: datetime.tzinfo | None) -> _Pipeline[_InT, datetime.datetime]
Transform a datetime value by replacing its timezone with the given value.
Unlike datetime_tz(), this does not validate the existing timezone, it
unconditionally sets tzinfo to tz.
_Pipeline[_InT, datetime.datetime]
tz : datetime.tzinfo | None
The timezone to attach to the datetime, or None to make it naive.
def str_lower() -> _Pipeline[_InT, str]
Transform a string value to lowercase.
_Pipeline[_InT, str]
def str_upper() -> _Pipeline[_InT, str]
Transform a string value to uppercase.
_Pipeline[_InT, str]
def str_title() -> _Pipeline[_InT, str]
Transform a string value to title case.
_Pipeline[_InT, str]
def str_strip() -> _Pipeline[_InT, str]
Strip leading and trailing whitespace from a string value.
_Pipeline[_InT, str]
def str_pattern(pattern: str) -> _Pipeline[_InT, str]
Constrain a string value to match a regular expression pattern.
_Pipeline[_InT, str]
pattern : str
The regular expression pattern the string must match.
def str_contains(substring: str) -> _Pipeline[_InT, str]
Constrain a string value to contain a given substring.
_Pipeline[_InT, str]
substring : str
The substring that must be present in the string.
def str_starts_with(prefix: str) -> _Pipeline[_InT, str]
Constrain a string value to start with a given prefix.
_Pipeline[_InT, str]
prefix : str
The prefix the string must start with.
def str_ends_with(suffix: str) -> _Pipeline[_InT, str]
Constrain a string value to end with a given suffix.
_Pipeline[_InT, str]
suffix : str
The suffix the string must end with.
def otherwise(
other: _Pipeline[_OtherIn, _OtherOut],
) -> _Pipeline[_InT | _OtherIn, _OutT | _OtherOut]
Combine two validation chains, returning the result of the first chain if it succeeds, and the second chain if it fails.
_Pipeline[_InT | _OtherIn, _OutT | _OtherOut]
def then(other: _Pipeline[_OutT, _OtherOut]) -> _Pipeline[_InT, _OtherOut]
Pipe the result of one validation chain into another.
_Pipeline[_InT, _OtherOut]
Experimental module exposing a function to generate a core schema that validates callable arguments.
def generate_arguments_schema(
func: Callable[..., Any],
schema_type: Literal['arguments', 'arguments-v3'] = 'arguments-v3',
parameters_callback: Callable[[int, str, Any], Literal['skip'] | None] | None = None,
config: ConfigDict | None = None,
) -> CoreSchema
Generate the schema for the arguments of a function.
CoreSchema — The generated schema.
The function to generate the schema for.
schema_type : Literal[‘arguments’, ‘arguments-v3’] Default: 'arguments-v3'
The type of schema to generate.
A callable that will be invoked for each parameter. The callback
should take three required arguments: the index, the name and the type annotation
(or Parameter.empty if not annotated) of the parameter.
The callback can optionally return 'skip', so that the parameter gets excluded
from the resulting schema.
config : ConfigDict | None Default: None
The configuration to use.