实验性 API

管道 API

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

_Pipeline dataclass

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

基类：Generic[_InT, _OutT]

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

transform

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

转换上一步的输出。

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

源代码在 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),))

参数模式 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]	要为其生成模式的函数。	必需
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)