Pipeline API

实验性 pipeline API 功能。请谨慎使用此 API，它可能会发生变化。

_Pipeline dataclass

_Pipeline(_steps: tuple[_Step, ...])

基类: Generic[_InT, _OutT]

验证、转换和解析步骤链的抽象表示。

transform

transform(
    func: Callable[[_OutT], _NewOutT]
) -> _Pipeline[_InT, _NewOutT]

转换上一步的输出。

如果用作 pipeline 中的第一步，则使用字段的类型。 也就是说，转换应用于将值解析为字段类型之后。

源代码位于 pydantic/experimental/pipeline.py

def transform(
    self,
    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.
    """
    return _Pipeline[_InT, _NewOutT](self._steps + (_Transform(func),))

validate_as

validate_as(
    tp: type[_NewOutT], *, strict: bool = ...
) -> _Pipeline[_InT, _NewOutT]

validate_as(
    tp: EllipsisType, *, strict: bool = ...
) -> _Pipeline[_InT, Any]

validate_as(
    tp: type[_NewOutT] | EllipsisType,
    *,
    strict: bool = False
) -> _Pipeline[_InT, Any]

验证/解析输入为新类型。

如果未提供类型，则使用字段的类型。

默认情况下，类型以 Pydantic 的 lax 模式解析，但您可以通过传递 strict=True 来启用 strict 模式。

源代码位于 pydantic/experimental/pipeline.py

def validate_as(self, tp: type[_NewOutT] | EllipsisType, *, strict: bool = False) -> _Pipeline[_InT, Any]:  # type: ignore
    """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`.
    """
    if isinstance(tp, EllipsisType):
        return _Pipeline[_InT, Any](self._steps + (_ValidateAs(_FieldTypeMarker, strict=strict),))
    return _Pipeline[_InT, _NewOutT](self._steps + (_ValidateAs(tp, strict=strict),))

validate_as_deferred

validate_as_deferred(
    func: Callable[[], type[_NewOutT]]
) -> _Pipeline[_InT, _NewOutT]

将输入解析为新类型，延迟类型的解析，直到当前类完全定义。

当您需要在其自身的类型注解中引用该类时，这非常有用。

源代码位于 pydantic/experimental/pipeline.py

def validate_as_deferred(self, 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.
    """
    return _Pipeline[_InT, _NewOutT](self._steps + (_ValidateAsDefer(func),))

constrain

constrain(constraint: Ge) -> _Pipeline[_InT, _NewOutGe]

constrain(constraint: Gt) -> _Pipeline[_InT, _NewOutGt]

constrain(constraint: Le) -> _Pipeline[_InT, _NewOutLe]

constrain(constraint: Lt) -> _Pipeline[_InT, _NewOutLt]

constrain(constraint: Len) -> _Pipeline[_InT, _NewOutLen]

constrain(
    constraint: MultipleOf,
) -> _Pipeline[_InT, _NewOutT]

constrain(
    constraint: Timezone,
) -> _Pipeline[_InT, _NewOutDatetime]

constrain(constraint: Predicate) -> _Pipeline[_InT, _OutT]

constrain(
    constraint: Interval,
) -> _Pipeline[_InT, _NewOutInterval]

constrain(constraint: _Eq) -> _Pipeline[_InT, _OutT]

constrain(constraint: _NotEq) -> _Pipeline[_InT, _OutT]

constrain(constraint: _In) -> _Pipeline[_InT, _OutT]

constrain(constraint: _NotIn) -> _Pipeline[_InT, _OutT]

constrain(
    constraint: Pattern[str],
) -> _Pipeline[_InT, _NewOutT]

constrain(constraint: _ConstraintAnnotation) -> Any

约束值以满足特定条件。

我们支持来自 annotated_types 的大多数条件，以及正则表达式。

大多数时候，您将调用像 gt、lt、len 等快捷方法，因此您无需直接调用此方法。

源代码位于 pydantic/experimental/pipeline.py

def constrain(self, constraint: _ConstraintAnnotation) -> Any:
    """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.
    """
    return _Pipeline[_InT, _OutT](self._steps + (_Constraint(constraint),))

predicate

predicate(
    func: Callable[[_NewOutT], bool]
) -> _Pipeline[_InT, _NewOutT]

约束值以满足特定谓词。

源代码位于 pydantic/experimental/pipeline.py

def predicate(self: _Pipeline[_InT, _NewOutT], func: Callable[[_NewOutT], bool]) -> _Pipeline[_InT, _NewOutT]:
    """Constrain a value to meet a certain predicate."""
    return self.constrain(annotated_types.Predicate(func))

gt

gt(gt: _NewOutGt) -> _Pipeline[_InT, _NewOutGt]

约束值必须大于某个值。

源代码位于 pydantic/experimental/pipeline.py

def gt(self: _Pipeline[_InT, _NewOutGt], gt: _NewOutGt) -> _Pipeline[_InT, _NewOutGt]:
    """Constrain a value to be greater than a certain value."""
    return self.constrain(annotated_types.Gt(gt))

lt

lt(lt: _NewOutLt) -> _Pipeline[_InT, _NewOutLt]

约束值必须小于某个值。

源代码位于 pydantic/experimental/pipeline.py

def lt(self: _Pipeline[_InT, _NewOutLt], lt: _NewOutLt) -> _Pipeline[_InT, _NewOutLt]:
    """Constrain a value to be less than a certain value."""
    return self.constrain(annotated_types.Lt(lt))

ge

ge(ge: _NewOutGe) -> _Pipeline[_InT, _NewOutGe]

约束值必须大于或等于某个值。

源代码位于 pydantic/experimental/pipeline.py

def ge(self: _Pipeline[_InT, _NewOutGe], ge: _NewOutGe) -> _Pipeline[_InT, _NewOutGe]:
    """Constrain a value to be greater than or equal to a certain value."""
    return self.constrain(annotated_types.Ge(ge))

le

le(le: _NewOutLe) -> _Pipeline[_InT, _NewOutLe]

约束值必须小于或等于某个值。

源代码位于 pydantic/experimental/pipeline.py

def le(self: _Pipeline[_InT, _NewOutLe], le: _NewOutLe) -> _Pipeline[_InT, _NewOutLe]:
    """Constrain a value to be less than or equal to a certain value."""
    return self.constrain(annotated_types.Le(le))

len

len(
    min_len: int, max_len: int | None = None
) -> _Pipeline[_InT, _NewOutLen]

约束值必须具有特定长度。

源代码位于 pydantic/experimental/pipeline.py

def len(self: _Pipeline[_InT, _NewOutLen], min_len: int, max_len: int | None = None) -> _Pipeline[_InT, _NewOutLen]:
    """Constrain a value to have a certain length."""
    return self.constrain(annotated_types.Len(min_len, max_len))

multiple_of

multiple_of(
    multiple_of: _NewOutDiv,
) -> _Pipeline[_InT, _NewOutDiv]

multiple_of(
    multiple_of: _NewOutMod,
) -> _Pipeline[_InT, _NewOutMod]

multiple_of(multiple_of: Any) -> _Pipeline[_InT, Any]

约束值必须是某个数字的倍数。

源代码位于 pydantic/experimental/pipeline.py

def multiple_of(self: _Pipeline[_InT, Any], multiple_of: Any) -> _Pipeline[_InT, Any]:
    """Constrain a value to be a multiple of a certain number."""
    return self.constrain(annotated_types.MultipleOf(multiple_of))

eq

eq(value: _OutT) -> _Pipeline[_InT, _OutT]

约束值必须等于某个值。

源代码位于 pydantic/experimental/pipeline.py

def eq(self: _Pipeline[_InT, _OutT], value: _OutT) -> _Pipeline[_InT, _OutT]:
    """Constrain a value to be equal to a certain value."""
    return self.constrain(_Eq(value))

not_eq

not_eq(value: _OutT) -> _Pipeline[_InT, _OutT]

约束值必须不等于某个值。

源代码位于 pydantic/experimental/pipeline.py

def not_eq(self: _Pipeline[_InT, _OutT], value: _OutT) -> _Pipeline[_InT, _OutT]:
    """Constrain a value to not be equal to a certain value."""
    return self.constrain(_NotEq(value))

in_

in_(values: Container[_OutT]) -> _Pipeline[_InT, _OutT]

约束值必须在某个集合中。

源代码位于 pydantic/experimental/pipeline.py

def in_(self: _Pipeline[_InT, _OutT], values: Container[_OutT]) -> _Pipeline[_InT, _OutT]:
    """Constrain a value to be in a certain set."""
    return self.constrain(_In(values))

not_in

not_in(values: Container[_OutT]) -> _Pipeline[_InT, _OutT]

约束值必须不在某个集合中。

源代码位于 pydantic/experimental/pipeline.py

def not_in(self: _Pipeline[_InT, _OutT], values: Container[_OutT]) -> _Pipeline[_InT, _OutT]:
    """Constrain a value to not be in a certain set."""
    return self.constrain(_NotIn(values))

otherwise

otherwise(
    other: _Pipeline[_OtherIn, _OtherOut]
) -> _Pipeline[_InT | _OtherIn, _OutT | _OtherOut]

组合两个验证链，如果第一个链成功，则返回第一个链的结果，如果失败，则返回第二个链的结果。

源代码位于 pydantic/experimental/pipeline.py

def otherwise(self, 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."""
    return _Pipeline((_PipelineOr(self, other),))

then

then(
    other: _Pipeline[_OutT, _OtherOut]
) -> _Pipeline[_InT, _OtherOut]

将一个验证链的结果管道输送到另一个验证链中。

源代码位于 pydantic/experimental/pipeline.py

def then(self, other: _Pipeline[_OutT, _OtherOut]) -> _Pipeline[_InT, _OtherOut]:
    """Pipe the result of one validation chain into another."""
    return _Pipeline((_PipelineAnd(self, other),))

Arguments schema API

实验性模块，公开了一个函数，用于生成验证可调用参数的核心模式。

generate_arguments_schema

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

为函数的参数生成模式。

参数

名称	类型	描述	默认值
func	Callable[..., Any]	要为其生成模式的函数。	required
schema_type	Literal['arguments', 'arguments-v3']	要生成的模式的类型。	'arguments-v3'
parameters_callback	Callable[[int, str, Any], Literal['skip'] | None] | None	将为每个参数调用的可调用对象。 回调应接受三个必需的参数：索引、名称和类型注解（或 Parameter.empty，如果未注解）。 回调可以选择返回 'skip'，以便将参数从结果模式中排除。	None
config	ConfigDict | None	要使用的配置。	None

返回

类型	描述
CoreSchema	生成的模式。

源代码位于 pydantic/experimental/arguments_schema.py

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.

    Args:
        func: The function to generate the schema for.
        schema_type: The type of schema to generate.
        parameters_callback: 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`][inspect.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: The configuration to use.

    Returns:
        The generated schema.
    """
    generate_schema = _generate_schema.GenerateSchema(
        _config.ConfigWrapper(config),
        ns_resolver=_namespace_utils.NsResolver(namespaces_tuple=_namespace_utils.ns_for_function(func)),
    )

    if schema_type == 'arguments':
        schema = generate_schema._arguments_schema(func, parameters_callback)  # pyright: ignore[reportArgumentType]
    else:
        schema = generate_schema._arguments_v3_schema(func, parameters_callback)  # pyright: ignore[reportArgumentType]
    return generate_schema.clean_schema(schema)