From cafb0ed1b72cbe7f1692761e5fbd33bfdf73de93 Mon Sep 17 00:00:00 2001 From: Polina Date: Thu, 2 Jul 2026 20:06:19 +0000 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=B8?= =?UTF-8?q?=D1=82=D1=8C=20=D1=84=D0=B0=D0=B9=D0=BB=D1=8B=20=D0=B2=20=C2=AB?= =?UTF-8?q?venv/Lib/site-packages/pydantic=C2=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- venv/Lib/site-packages/pydantic/root_model.py | 157 + venv/Lib/site-packages/pydantic/schema.py | 5 + venv/Lib/site-packages/pydantic/tools.py | 5 + .../site-packages/pydantic/type_adapter.py | 801 ++++ venv/Lib/site-packages/pydantic/types.py | 3310 +++++++++++++++++ 5 files changed, 4278 insertions(+) create mode 100644 venv/Lib/site-packages/pydantic/root_model.py create mode 100644 venv/Lib/site-packages/pydantic/schema.py create mode 100644 venv/Lib/site-packages/pydantic/tools.py create mode 100644 venv/Lib/site-packages/pydantic/type_adapter.py create mode 100644 venv/Lib/site-packages/pydantic/types.py diff --git a/venv/Lib/site-packages/pydantic/root_model.py b/venv/Lib/site-packages/pydantic/root_model.py new file mode 100644 index 0000000..d290043 --- /dev/null +++ b/venv/Lib/site-packages/pydantic/root_model.py @@ -0,0 +1,157 @@ +"""RootModel class and type definitions.""" + +from __future__ import annotations as _annotations + +from copy import copy, deepcopy +from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar + +from pydantic_core import PydanticUndefined +from typing_extensions import Self, dataclass_transform + +from . import PydanticUserError +from ._internal import _model_construction, _repr +from .main import BaseModel, _object_setattr + +if TYPE_CHECKING: + from .fields import Field as PydanticModelField + from .fields import PrivateAttr as PydanticModelPrivateAttr + + # dataclass_transform could be applied to RootModel directly, but `ModelMetaclass`'s dataclass_transform + # takes priority (at least with pyright). We trick type checkers into thinking we apply dataclass_transform + # on a new metaclass. + @dataclass_transform(kw_only_default=False, field_specifiers=(PydanticModelField, PydanticModelPrivateAttr)) + class _RootModelMetaclass(_model_construction.ModelMetaclass): ... +else: + _RootModelMetaclass = _model_construction.ModelMetaclass + +__all__ = ('RootModel',) + +RootModelRootType = TypeVar('RootModelRootType') + + +class RootModel(BaseModel, Generic[RootModelRootType], metaclass=_RootModelMetaclass): + """!!! abstract "Usage Documentation" + [`RootModel` and Custom Root Types](../concepts/models.md#rootmodel-and-custom-root-types) + + A Pydantic `BaseModel` for the root object of the model. + + Attributes: + root: The root object of the model. + __pydantic_root_model__: Whether the model is a RootModel. + __pydantic_private__: Private fields in the model. + __pydantic_extra__: Extra fields in the model. + + """ + + __pydantic_root_model__ = True + __pydantic_private__ = None + __pydantic_extra__ = None + + root: RootModelRootType + + def __init_subclass__(cls, **kwargs): + extra = cls.model_config.get('extra') + if extra is not None: + raise PydanticUserError( + "`RootModel` does not support setting `model_config['extra']`", code='root-model-extra' + ) + super().__init_subclass__(**kwargs) + + def __init__(self, /, root: RootModelRootType = PydanticUndefined, **data) -> None: # type: ignore + __tracebackhide__ = True + if data: + if root is not PydanticUndefined: + raise ValueError( + '"RootModel.__init__" accepts either a single positional argument or arbitrary keyword arguments' + ) + root = data # type: ignore + self.__pydantic_validator__.validate_python(root, self_instance=self) + + __init__.__pydantic_base_init__ = True # pyright: ignore[reportFunctionMemberAccess] + + @classmethod + def model_construct(cls, root: RootModelRootType, _fields_set: set[str] | None = None) -> Self: # type: ignore + """Create a new model using the provided root object and update fields set. + + Args: + root: The root object of the model. + _fields_set: The set of fields to be updated. + + Returns: + The new model. + + Raises: + NotImplemented: If the model is not a subclass of `RootModel`. + """ + return super().model_construct(root=root, _fields_set=_fields_set) + + def __getstate__(self) -> dict[Any, Any]: + return { + '__dict__': self.__dict__, + '__pydantic_fields_set__': self.__pydantic_fields_set__, + } + + def __setstate__(self, state: dict[Any, Any]) -> None: + _object_setattr(self, '__pydantic_fields_set__', state['__pydantic_fields_set__']) + _object_setattr(self, '__dict__', state['__dict__']) + + def __copy__(self) -> Self: + """Returns a shallow copy of the model.""" + cls = type(self) + m = cls.__new__(cls) + new_dict = copy(self.__dict__) + new_dict['root'] = copy(self.__dict__['root']) + _object_setattr(m, '__dict__', new_dict) + _object_setattr(m, '__pydantic_fields_set__', copy(self.__pydantic_fields_set__)) + return m + + def __deepcopy__(self, memo: dict[int, Any] | None = None) -> Self: + """Returns a deep copy of the model.""" + cls = type(self) + m = cls.__new__(cls) + _object_setattr(m, '__dict__', deepcopy(self.__dict__, memo=memo)) + # This next line doesn't need a deepcopy because __pydantic_fields_set__ is a set[str], + # and attempting a deepcopy would be marginally slower. + _object_setattr(m, '__pydantic_fields_set__', copy(self.__pydantic_fields_set__)) + return m + + if TYPE_CHECKING: + + def model_dump( # type: ignore + self, + *, + mode: Literal['json', 'python'] | str = 'python', + include: Any = None, + exclude: Any = None, + context: dict[str, Any] | None = None, + by_alias: bool | None = None, + exclude_unset: bool = False, + exclude_defaults: bool = False, + exclude_none: bool = False, + exclude_computed_fields: bool = False, + round_trip: bool = False, + warnings: bool | Literal['none', 'warn', 'error'] = True, + serialize_as_any: bool = False, + ) -> Any: + """This method is included just to get a more accurate return type for type checkers. + It is included in this `if TYPE_CHECKING:` block since no override is actually necessary. + + See the documentation of `BaseModel.model_dump` for more details about the arguments. + + Generally, this method will have a return type of `RootModelRootType`, assuming that `RootModelRootType` is + not a `BaseModel` subclass. If `RootModelRootType` is a `BaseModel` subclass, then the return + type will likely be `dict[str, Any]`, as `model_dump` calls are recursive. The return type could + even be something different, in the case of a custom serializer. + Thus, `Any` is used here to catch all of these cases. + """ + ... + + def __eq__(self, other: Any) -> bool: + if not isinstance(other, RootModel): + return NotImplemented + return self.__pydantic_fields__['root'].annotation == other.__pydantic_fields__[ + 'root' + ].annotation and super().__eq__(other) + + def __repr_args__(self) -> _repr.ReprArgs: + yield 'root', self.root diff --git a/venv/Lib/site-packages/pydantic/schema.py b/venv/Lib/site-packages/pydantic/schema.py new file mode 100644 index 0000000..a3245a6 --- /dev/null +++ b/venv/Lib/site-packages/pydantic/schema.py @@ -0,0 +1,5 @@ +"""The `schema` module is a backport module from V1.""" + +from ._migration import getattr_migration + +__getattr__ = getattr_migration(__name__) diff --git a/venv/Lib/site-packages/pydantic/tools.py b/venv/Lib/site-packages/pydantic/tools.py new file mode 100644 index 0000000..fdc68c4 --- /dev/null +++ b/venv/Lib/site-packages/pydantic/tools.py @@ -0,0 +1,5 @@ +"""The `tools` module is a backport module from V1.""" + +from ._migration import getattr_migration + +__getattr__ = getattr_migration(__name__) diff --git a/venv/Lib/site-packages/pydantic/type_adapter.py b/venv/Lib/site-packages/pydantic/type_adapter.py new file mode 100644 index 0000000..d962305 --- /dev/null +++ b/venv/Lib/site-packages/pydantic/type_adapter.py @@ -0,0 +1,801 @@ +"""Type adapter specification.""" + +from __future__ import annotations as _annotations + +import sys +import types +from collections.abc import Callable, Iterable +from dataclasses import is_dataclass +from types import FrameType +from typing import ( + Any, + Generic, + Literal, + TypeVar, + cast, + final, + overload, +) + +from pydantic_core import CoreSchema, SchemaSerializer, SchemaValidator, Some +from typing_extensions import ParamSpec, is_typeddict + +from pydantic.errors import PydanticUserError +from pydantic.main import BaseModel, IncEx + +from ._internal import _config, _generate_schema, _mock_val_ser, _namespace_utils, _repr, _typing_extra, _utils +from .config import ConfigDict, ExtraValues +from .errors import PydanticUndefinedAnnotation +from .json_schema import ( + DEFAULT_REF_TEMPLATE, + GenerateJsonSchema, + JsonSchemaKeyT, + JsonSchemaMode, + JsonSchemaValue, +) +from .plugin._schema_validator import PluggableSchemaValidator, create_schema_validator + +T = TypeVar('T') +R = TypeVar('R') +P = ParamSpec('P') +TypeAdapterT = TypeVar('TypeAdapterT', bound='TypeAdapter') + + +def _getattr_no_parents(obj: Any, attribute: str) -> Any: + """Returns the attribute value without attempting to look up attributes from parent types.""" + if hasattr(obj, '__dict__'): + try: + return obj.__dict__[attribute] + except KeyError: + pass + + slots = getattr(obj, '__slots__', None) + if slots is not None and attribute in slots: + return getattr(obj, attribute) + else: + raise AttributeError(attribute) + + +def _type_has_config(type_: Any) -> bool: + """Returns whether the type has config.""" + type_ = _typing_extra.annotated_type(type_) or type_ + try: + return issubclass(type_, BaseModel) or is_dataclass(type_) or is_typeddict(type_) + except TypeError: + # type is not a class + return False + + +@final +class TypeAdapter(Generic[T]): + """!!! abstract "Usage Documentation" + [`TypeAdapter`](../concepts/type_adapter.md) + + Type adapters provide a flexible way to perform validation and serialization based on a Python type. + + A `TypeAdapter` instance exposes some of the functionality from `BaseModel` instance methods + for types that do not have such methods (such as dataclasses, primitive types, and more). + + **Note:** `TypeAdapter` instances are not types, and cannot be used as type annotations for fields. + + Args: + type: The type associated with the `TypeAdapter`. + config: Configuration for the `TypeAdapter`, should be a dictionary conforming to + [`ConfigDict`][pydantic.config.ConfigDict]. + + !!! note + You cannot provide a configuration when instantiating a `TypeAdapter` if the type you're using + has its own config that cannot be overridden (ex: `BaseModel`, `TypedDict`, and `dataclass`). A + [`type-adapter-config-unused`](../errors/usage_errors.md#type-adapter-config-unused) error will + be raised in this case. + _parent_depth: Depth at which to search for the [parent frame][frame-objects]. This frame is used when + resolving forward annotations during schema building, by looking for the globals and locals of this + frame. Defaults to 2, which will result in the frame where the `TypeAdapter` was instantiated. + + !!! note + This parameter is named with an underscore to suggest its private nature and discourage use. + It may be deprecated in a minor version, so we only recommend using it if you're comfortable + with potential change in behavior/support. It's default value is 2 because internally, + the `TypeAdapter` class makes another call to fetch the frame. + module: The module that passes to plugin if provided. + + Attributes: + core_schema: The core schema for the type. + validator: The schema validator for the type. + serializer: The schema serializer for the type. + pydantic_complete: Whether the core schema for the type is successfully built. + + ??? tip "Compatibility with `mypy`" + Depending on the type used, `mypy` might raise an error when instantiating a `TypeAdapter`. As a workaround, you can explicitly + annotate your variable: + + ```py + from typing import Union + + from pydantic import TypeAdapter + + ta: TypeAdapter[Union[str, int]] = TypeAdapter(Union[str, int]) # type: ignore[arg-type] + ``` + + ??? info "Namespace management nuances and implementation details" + + Here, we collect some notes on namespace management, and subtle differences from `BaseModel`: + + `BaseModel` uses its own `__module__` to find out where it was defined + and then looks for symbols to resolve forward references in those globals. + On the other hand, `TypeAdapter` can be initialized with arbitrary objects, + which may not be types and thus do not have a `__module__` available. + So instead we look at the globals in our parent stack frame. + + It is expected that the `ns_resolver` passed to this function will have the correct + namespace for the type we're adapting. See the source code for `TypeAdapter.__init__` + and `TypeAdapter.rebuild` for various ways to construct this namespace. + + This works for the case where this function is called in a module that + has the target of forward references in its scope, but + does not always work for more complex cases. + + For example, take the following: + + ```python {title="a.py"} + IntList = list[int] + OuterDict = dict[str, 'IntList'] + ``` + + ```python {test="skip" title="b.py"} + from a import OuterDict + + from pydantic import TypeAdapter + + IntList = int # replaces the symbol the forward reference is looking for + v = TypeAdapter(OuterDict) + v({'x': 1}) # should fail but doesn't + ``` + + If `OuterDict` were a `BaseModel`, this would work because it would resolve + the forward reference within the `a.py` namespace. + But `TypeAdapter(OuterDict)` can't determine what module `OuterDict` came from. + + In other words, the assumption that _all_ forward references exist in the + module we are being called from is not technically always true. + Although most of the time it is and it works fine for recursive models and such, + `BaseModel`'s behavior isn't perfect either and _can_ break in similar ways, + so there is no right or wrong between the two. + + But at the very least this behavior is _subtly_ different from `BaseModel`'s. + """ + + core_schema: CoreSchema + validator: SchemaValidator | PluggableSchemaValidator + serializer: SchemaSerializer + pydantic_complete: bool + + @overload + def __init__( + self, + type: type[T], + *, + config: ConfigDict | None = ..., + _parent_depth: int = ..., + module: str | None = ..., + ) -> None: ... + + # This second overload is for unsupported special forms (such as Annotated, Union, etc.) + # Currently there is no way to type this correctly + # See https://github.com/python/typing/pull/1618 + @overload + def __init__( + self, + type: Any, + *, + config: ConfigDict | None = ..., + _parent_depth: int = ..., + module: str | None = ..., + ) -> None: ... + + def __init__( + self, + type: Any, + *, + config: ConfigDict | None = None, + _parent_depth: int = 2, + module: str | None = None, + ) -> None: + if _type_has_config(type) and config is not None: + raise PydanticUserError( + 'Cannot use `config` when the type is a BaseModel, dataclass or TypedDict.' + ' These types can have their own config and setting the config via the `config`' + ' parameter to TypeAdapter will not override it, thus the `config` you passed to' + ' TypeAdapter becomes meaningless, which is probably not what you want.', + code='type-adapter-config-unused', + ) + + self._type = type + self._config = config + self._parent_depth = _parent_depth + self.pydantic_complete = False + + parent_frame = self._fetch_parent_frame() + if isinstance(type, types.FunctionType): + # Special case functions, which are *not* pushed to the `NsResolver` stack and without this special case + # would only have access to the parent namespace where the `TypeAdapter` was instantiated (if the function is defined + # in another module, we need to look at that module's globals). + if parent_frame is not None: + # `f_locals` is the namespace where the type adapter was instantiated (~ to `f_globals` if at the module level): + parent_ns = parent_frame.f_locals + else: # pragma: no cover + parent_ns = None + globalns, localns = _namespace_utils.ns_for_function( + type, + parent_namespace=parent_ns, + ) + parent_namespace = None + else: + if parent_frame is not None: + globalns = parent_frame.f_globals + # Do not provide a local ns if the type adapter happens to be instantiated at the module level: + localns = parent_frame.f_locals if parent_frame.f_locals is not globalns else {} + else: # pragma: no cover + globalns = {} + localns = {} + parent_namespace = localns + + self._module_name = module or cast(str, globalns.get('__name__', '')) + self._init_core_attrs( + ns_resolver=_namespace_utils.NsResolver( + namespaces_tuple=_namespace_utils.NamespacesTuple(locals=localns, globals=globalns), + parent_namespace=parent_namespace, + ), + force=False, + ) + + def _fetch_parent_frame(self) -> FrameType | None: + frame = sys._getframe(self._parent_depth) + if frame.f_globals.get('__name__') == 'typing': + # Because `TypeAdapter` is generic, explicitly parametrizing the class results + # in a `typing._GenericAlias` instance, which proxies instantiation calls to the + # "real" `TypeAdapter` class and thus adding an extra frame to the call. To avoid + # pulling anything from the `typing` module, use the correct frame (the one before): + return frame.f_back + + return frame + + def _init_core_attrs( + self, ns_resolver: _namespace_utils.NsResolver, force: bool, raise_errors: bool = False + ) -> bool: + """Initialize the core schema, validator, and serializer for the type. + + Args: + ns_resolver: The namespace resolver to use when building the core schema for the adapted type. + force: Whether to force the construction of the core schema, validator, and serializer. + If `force` is set to `False` and `_defer_build` is `True`, the core schema, validator, and serializer will be set to mocks. + raise_errors: Whether to raise errors if initializing any of the core attrs fails. + + Returns: + `True` if the core schema, validator, and serializer were successfully initialized, otherwise `False`. + + Raises: + PydanticUndefinedAnnotation: If `PydanticUndefinedAnnotation` occurs in`__get_pydantic_core_schema__` + and `raise_errors=True`. + """ + if not force and self._defer_build: + _mock_val_ser.set_type_adapter_mocks(self) + self.pydantic_complete = False + return False + + try: + self.core_schema = _getattr_no_parents(self._type, '__pydantic_core_schema__') + self.validator = _getattr_no_parents(self._type, '__pydantic_validator__') + self.serializer = _getattr_no_parents(self._type, '__pydantic_serializer__') + + # TODO: we don't go through the rebuild logic here directly because we don't want + # to repeat all of the namespace fetching logic that we've already done + # so we simply skip to the block below that does the actual schema generation + if ( + isinstance(self.core_schema, _mock_val_ser.MockCoreSchema) + or isinstance(self.validator, _mock_val_ser.MockValSer) + or isinstance(self.serializer, _mock_val_ser.MockValSer) + ): + raise AttributeError() + except AttributeError: + config_wrapper = _config.ConfigWrapper(self._config) + + schema_generator = _generate_schema.GenerateSchema(config_wrapper, ns_resolver=ns_resolver) + + try: + core_schema = schema_generator.generate_schema(self._type) + except PydanticUndefinedAnnotation: + if raise_errors: + raise + _mock_val_ser.set_type_adapter_mocks(self) + return False + + try: + self.core_schema = schema_generator.clean_schema(core_schema) + except _generate_schema.InvalidSchemaError: + _mock_val_ser.set_type_adapter_mocks(self) + return False + + core_config = config_wrapper.core_config(None) + + self.validator = create_schema_validator( + schema=self.core_schema, + schema_type=self._type, + schema_type_module=self._module_name, + schema_type_name=str(self._type), + schema_kind='TypeAdapter', + config=core_config, + plugin_settings=config_wrapper.plugin_settings, + ) + self.serializer = SchemaSerializer(self.core_schema, core_config) + + self.pydantic_complete = True + return True + + @property + def _defer_build(self) -> bool: + config = self._config if self._config is not None else self._model_config + if config: + return config.get('defer_build') is True + return False + + @property + def _model_config(self) -> ConfigDict | None: + type_: Any = _typing_extra.annotated_type(self._type) or self._type # Eg FastAPI heavily uses Annotated + if _utils.lenient_issubclass(type_, BaseModel): + return type_.model_config + return getattr(type_, '__pydantic_config__', None) + + def __repr__(self) -> str: + return f'TypeAdapter({_repr.display_as_type(self._type)})' + + def rebuild( + self, + *, + force: bool = False, + raise_errors: bool = True, + _parent_namespace_depth: int = 2, + _types_namespace: _namespace_utils.MappingNamespace | None = None, + ) -> bool | None: + """Try to rebuild the pydantic-core schema for the adapter's type. + + This may be necessary when one of the annotations is a ForwardRef which could not be resolved during + the initial attempt to build the schema, and automatic rebuilding fails. + + Args: + force: Whether to force the rebuilding of the type adapter's schema, defaults to `False`. + raise_errors: Whether to raise errors, defaults to `True`. + _parent_namespace_depth: Depth at which to search for the [parent frame][frame-objects]. This + frame is used when resolving forward annotations during schema rebuilding, by looking for + the locals of this frame. Defaults to 2, which will result in the frame where the method + was called. + _types_namespace: An explicit types namespace to use, instead of using the local namespace + from the parent frame. Defaults to `None`. + + Returns: + Returns `None` if the schema is already "complete" and rebuilding was not required. + If rebuilding _was_ required, returns `True` if rebuilding was successful, otherwise `False`. + """ + if not force and self.pydantic_complete: + return None + + if _types_namespace is not None: + rebuild_ns = _types_namespace + elif _parent_namespace_depth > 0: + rebuild_ns = _typing_extra.parent_frame_namespace(parent_depth=_parent_namespace_depth, force=True) or {} + else: + rebuild_ns = {} + + # we have to manually fetch globals here because there's no type on the stack of the NsResolver + # and so we skip the globalns = get_module_ns_of(typ) call that would normally happen + globalns = sys._getframe(max(_parent_namespace_depth - 1, 1)).f_globals + ns_resolver = _namespace_utils.NsResolver( + namespaces_tuple=_namespace_utils.NamespacesTuple(locals=rebuild_ns, globals=globalns), + parent_namespace=rebuild_ns, + ) + return self._init_core_attrs(ns_resolver=ns_resolver, force=True, raise_errors=raise_errors) + + def validate_python( + self, + object: Any, + /, + *, + strict: bool | None = None, + extra: ExtraValues | None = None, + from_attributes: bool | None = None, + context: Any | None = None, + experimental_allow_partial: bool | Literal['off', 'on', 'trailing-strings'] = False, + by_alias: bool | None = None, + by_name: bool | None = None, + ) -> T: + """Validate a Python object against the model. + + Args: + object: The Python object to validate against the model. + strict: Whether to strictly check types. + extra: Whether to ignore, allow, or forbid extra data during model validation. + See the [`extra` configuration value][pydantic.ConfigDict.extra] for details. + from_attributes: Whether to extract data from object attributes. + context: Additional context to pass to the validator. + experimental_allow_partial: **Experimental** whether to enable + [partial validation](../concepts/experimental.md#partial-validation), e.g. to process streams. + * False / 'off': Default behavior, no partial validation. + * True / 'on': Enable partial validation. + * 'trailing-strings': Enable partial validation and allow trailing strings in the input. + by_alias: Whether to use the field's alias when validating against the provided input data. + by_name: Whether to use the field's name when validating against the provided input data. + + !!! note + When using `TypeAdapter` with a Pydantic `dataclass`, the use of the `from_attributes` + argument is not supported. + + Returns: + The validated object. + """ + if by_alias is False and by_name is not True: + raise PydanticUserError( + 'At least one of `by_alias` or `by_name` must be set to True.', + code='validate-by-alias-and-name-false', + ) + + return self.validator.validate_python( + object, + strict=strict, + extra=extra, + from_attributes=from_attributes, + context=context, + allow_partial=experimental_allow_partial, + by_alias=by_alias, + by_name=by_name, + ) + + def validate_json( + self, + data: str | bytes | bytearray, + /, + *, + strict: bool | None = None, + extra: ExtraValues | None = None, + context: Any | None = None, + experimental_allow_partial: bool | Literal['off', 'on', 'trailing-strings'] = False, + by_alias: bool | None = None, + by_name: bool | None = None, + ) -> T: + """!!! abstract "Usage Documentation" + [JSON Parsing](../concepts/json.md#json-parsing) + + Validate a JSON string or bytes against the model. + + Args: + data: The JSON data to validate against the model. + strict: Whether to strictly check types. + extra: Whether to ignore, allow, or forbid extra data during model validation. + See the [`extra` configuration value][pydantic.ConfigDict.extra] for details. + context: Additional context to use during validation. + experimental_allow_partial: **Experimental** whether to enable + [partial validation](../concepts/experimental.md#partial-validation), e.g. to process streams. + * False / 'off': Default behavior, no partial validation. + * True / 'on': Enable partial validation. + * 'trailing-strings': Enable partial validation and allow trailing strings in the input. + by_alias: Whether to use the field's alias when validating against the provided input data. + by_name: Whether to use the field's name when validating against the provided input data. + + Returns: + The validated object. + """ + if by_alias is False and by_name is not True: + raise PydanticUserError( + 'At least one of `by_alias` or `by_name` must be set to True.', + code='validate-by-alias-and-name-false', + ) + + return self.validator.validate_json( + data, + strict=strict, + extra=extra, + context=context, + allow_partial=experimental_allow_partial, + by_alias=by_alias, + by_name=by_name, + ) + + def validate_strings( + self, + obj: Any, + /, + *, + strict: bool | None = None, + extra: ExtraValues | None = None, + context: Any | None = None, + experimental_allow_partial: bool | Literal['off', 'on', 'trailing-strings'] = False, + by_alias: bool | None = None, + by_name: bool | None = None, + ) -> T: + """Validate object contains string data against the model. + + Args: + obj: The object contains string data to validate. + strict: Whether to strictly check types. + extra: Whether to ignore, allow, or forbid extra data during model validation. + See the [`extra` configuration value][pydantic.ConfigDict.extra] for details. + context: Additional context to use during validation. + experimental_allow_partial: **Experimental** whether to enable + [partial validation](../concepts/experimental.md#partial-validation), e.g. to process streams. + * False / 'off': Default behavior, no partial validation. + * True / 'on': Enable partial validation. + * 'trailing-strings': Enable partial validation and allow trailing strings in the input. + by_alias: Whether to use the field's alias when validating against the provided input data. + by_name: Whether to use the field's name when validating against the provided input data. + + Returns: + The validated object. + """ + if by_alias is False and by_name is not True: + raise PydanticUserError( + 'At least one of `by_alias` or `by_name` must be set to True.', + code='validate-by-alias-and-name-false', + ) + + return self.validator.validate_strings( + obj, + strict=strict, + extra=extra, + context=context, + allow_partial=experimental_allow_partial, + by_alias=by_alias, + by_name=by_name, + ) + + def get_default_value(self, *, strict: bool | None = None, context: Any | None = None) -> Some[T] | None: + """Get the default value for the wrapped type. + + Args: + strict: Whether to strictly check types. + context: Additional context to pass to the validator. + + Returns: + The default value wrapped in a `Some` if there is one or None if not. + """ + return self.validator.get_default_value(strict=strict, context=context) + + def dump_python( + self, + instance: T, + /, + *, + mode: Literal['json', 'python'] = 'python', + include: IncEx | None = None, + exclude: IncEx | None = None, + by_alias: bool | None = None, + exclude_unset: bool = False, + exclude_defaults: bool = False, + exclude_none: bool = False, + exclude_computed_fields: bool = False, + round_trip: bool = False, + warnings: bool | Literal['none', 'warn', 'error'] = True, + fallback: Callable[[Any], Any] | None = None, + serialize_as_any: bool = False, + polymorphic_serialization: bool | None = None, + context: Any | None = None, + ) -> Any: + """Dump an instance of the adapted type to a Python object. + + Args: + instance: The Python object to serialize. + mode: The output format. + include: Fields to include in the output. + exclude: Fields to exclude from the output. + by_alias: Whether to use alias names for field names. + exclude_unset: Whether to exclude unset fields. + exclude_defaults: Whether to exclude fields with default values. + exclude_none: Whether to exclude fields with None values. + exclude_computed_fields: Whether to exclude computed fields. + While this can be useful for round-tripping, it is usually recommended to use the dedicated + `round_trip` parameter instead. + round_trip: Whether to output the serialized data in a way that is compatible with deserialization. + warnings: How to handle serialization errors. False/"none" ignores them, True/"warn" logs errors, + "error" raises a [`PydanticSerializationError`][pydantic_core.PydanticSerializationError]. + fallback: A function to call when an unknown value is encountered. If not provided, + a [`PydanticSerializationError`][pydantic_core.PydanticSerializationError] error is raised. + serialize_as_any: Whether to serialize fields with duck-typing serialization behavior. + polymorphic_serialization: Whether to use model and dataclass polymorphic serialization for this call. + context: Additional context to pass to the serializer. + + Returns: + The serialized object. + """ + return self.serializer.to_python( + instance, + mode=mode, + by_alias=by_alias, + include=include, + exclude=exclude, + exclude_unset=exclude_unset, + exclude_defaults=exclude_defaults, + exclude_none=exclude_none, + exclude_computed_fields=exclude_computed_fields, + round_trip=round_trip, + warnings=warnings, + fallback=fallback, + serialize_as_any=serialize_as_any, + polymorphic_serialization=polymorphic_serialization, + context=context, + ) + + def dump_json( + self, + instance: T, + /, + *, + indent: int | None = None, + ensure_ascii: bool = False, + include: IncEx | None = None, + exclude: IncEx | None = None, + by_alias: bool | None = None, + exclude_unset: bool = False, + exclude_defaults: bool = False, + exclude_none: bool = False, + exclude_computed_fields: bool = False, + round_trip: bool = False, + warnings: bool | Literal['none', 'warn', 'error'] = True, + fallback: Callable[[Any], Any] | None = None, + serialize_as_any: bool = False, + polymorphic_serialization: bool | None = None, + context: Any | None = None, + ) -> bytes: + """!!! abstract "Usage Documentation" + [JSON Serialization](../concepts/json.md#json-serialization) + + Serialize an instance of the adapted type to JSON. + + Args: + instance: The instance to be serialized. + indent: Number of spaces for JSON indentation. + ensure_ascii: If `True`, the output is guaranteed to have all incoming non-ASCII characters escaped. + If `False` (the default), these characters will be output as-is. + include: Fields to include. + exclude: Fields to exclude. + by_alias: Whether to use alias names for field names. + exclude_unset: Whether to exclude unset fields. + exclude_defaults: Whether to exclude fields with default values. + exclude_none: Whether to exclude fields with a value of `None`. + exclude_computed_fields: Whether to exclude computed fields. + While this can be useful for round-tripping, it is usually recommended to use the dedicated + `round_trip` parameter instead. + round_trip: Whether to serialize and deserialize the instance to ensure round-tripping. + warnings: How to handle serialization errors. False/"none" ignores them, True/"warn" logs errors, + "error" raises a [`PydanticSerializationError`][pydantic_core.PydanticSerializationError]. + fallback: A function to call when an unknown value is encountered. If not provided, + a [`PydanticSerializationError`][pydantic_core.PydanticSerializationError] error is raised. + serialize_as_any: Whether to serialize fields with duck-typing serialization behavior. + polymorphic_serialization: Whether to use model and dataclass polymorphic serialization for this call. + context: Additional context to pass to the serializer. + + Returns: + The JSON representation of the given instance as bytes. + """ + return self.serializer.to_json( + instance, + indent=indent, + ensure_ascii=ensure_ascii, + include=include, + exclude=exclude, + by_alias=by_alias, + exclude_unset=exclude_unset, + exclude_defaults=exclude_defaults, + exclude_none=exclude_none, + exclude_computed_fields=exclude_computed_fields, + round_trip=round_trip, + warnings=warnings, + fallback=fallback, + serialize_as_any=serialize_as_any, + polymorphic_serialization=polymorphic_serialization, + context=context, + ) + + def json_schema( + self, + *, + by_alias: bool = True, + ref_template: str = DEFAULT_REF_TEMPLATE, + union_format: Literal['any_of', 'primitive_type_array'] = 'any_of', + schema_generator: type[GenerateJsonSchema] = GenerateJsonSchema, + mode: JsonSchemaMode = 'validation', + ) -> dict[str, Any]: + """Generate a JSON schema for the adapted type. + + Args: + by_alias: Whether to use alias names for field names. + ref_template: The format string used for generating $ref strings. + union_format: The format to use when combining schemas from unions together. Can be one of: + + - `'any_of'`: Use the [`anyOf`](https://json-schema.org/understanding-json-schema/reference/combining#anyOf) + keyword to combine schemas (the default). + - `'primitive_type_array'`: Use the [`type`](https://json-schema.org/understanding-json-schema/reference/type) + keyword as an array of strings, containing each type of the combination. If any of the schemas is not a primitive + type (`string`, `boolean`, `null`, `integer` or `number`) or contains constraints/metadata, falls back to + `any_of`. + schema_generator: To override the logic used to generate the JSON schema, as a subclass of + `GenerateJsonSchema` with your desired modifications + mode: The mode in which to generate the schema. + schema_generator: The generator class used for creating the schema. + mode: The mode to use for schema generation. + + Returns: + The JSON schema for the model as a dictionary. + """ + schema_generator_instance = schema_generator( + by_alias=by_alias, ref_template=ref_template, union_format=union_format + ) + if isinstance(self.core_schema, _mock_val_ser.MockCoreSchema): + self.core_schema.rebuild() + assert not isinstance(self.core_schema, _mock_val_ser.MockCoreSchema), 'this is a bug! please report it' + return schema_generator_instance.generate(self.core_schema, mode=mode) + + @staticmethod + def json_schemas( + inputs: Iterable[tuple[JsonSchemaKeyT, JsonSchemaMode, TypeAdapter[Any]]], + /, + *, + by_alias: bool = True, + title: str | None = None, + description: str | None = None, + ref_template: str = DEFAULT_REF_TEMPLATE, + union_format: Literal['any_of', 'primitive_type_array'] = 'any_of', + schema_generator: type[GenerateJsonSchema] = GenerateJsonSchema, + ) -> tuple[dict[tuple[JsonSchemaKeyT, JsonSchemaMode], JsonSchemaValue], JsonSchemaValue]: + """Generate a JSON schema including definitions from multiple type adapters. + + Args: + inputs: Inputs to schema generation. The first two items will form the keys of the (first) + output mapping; the type adapters will provide the core schemas that get converted into + definitions in the output JSON schema. + by_alias: Whether to use alias names. + title: The title for the schema. + description: The description for the schema. + ref_template: The format string used for generating $ref strings. + union_format: The format to use when combining schemas from unions together. Can be one of: + + - `'any_of'`: Use the [`anyOf`](https://json-schema.org/understanding-json-schema/reference/combining#anyOf) + keyword to combine schemas (the default). + - `'primitive_type_array'`: Use the [`type`](https://json-schema.org/understanding-json-schema/reference/type) + keyword as an array of strings, containing each type of the combination. If any of the schemas is not a primitive + type (`string`, `boolean`, `null`, `integer` or `number`) or contains constraints/metadata, falls back to + `any_of`. + schema_generator: The generator class used for creating the schema. + + Returns: + A tuple where: + + - The first element is a dictionary whose keys are tuples of JSON schema key type and JSON mode, and + whose values are the JSON schema corresponding to that pair of inputs. (These schemas may have + JsonRef references to definitions that are defined in the second returned element.) + - The second element is a JSON schema containing all definitions referenced in the first returned + element, along with the optional title and description keys. + + """ + schema_generator_instance = schema_generator( + by_alias=by_alias, ref_template=ref_template, union_format=union_format + ) + + inputs_ = [] + for key, mode, adapter in inputs: + # This is the same pattern we follow for model json schemas - we attempt a core schema rebuild if we detect a mock + if isinstance(adapter.core_schema, _mock_val_ser.MockCoreSchema): + adapter.core_schema.rebuild() + assert not isinstance(adapter.core_schema, _mock_val_ser.MockCoreSchema), ( + 'this is a bug! please report it' + ) + inputs_.append((key, mode, adapter.core_schema)) + + json_schemas_map, definitions = schema_generator_instance.generate_definitions(inputs_) + + json_schema: dict[str, Any] = {} + if definitions: + json_schema['$defs'] = definitions + if title: + json_schema['title'] = title + if description: + json_schema['description'] = description + + return json_schemas_map, json_schema diff --git a/venv/Lib/site-packages/pydantic/types.py b/venv/Lib/site-packages/pydantic/types.py new file mode 100644 index 0000000..467bc82 --- /dev/null +++ b/venv/Lib/site-packages/pydantic/types.py @@ -0,0 +1,3310 @@ +"""The types module contains custom types used by pydantic.""" + +from __future__ import annotations as _annotations + +import base64 +import dataclasses as _dataclasses +import re +from collections.abc import Hashable, Iterator +from datetime import date, datetime +from decimal import Decimal +from enum import Enum +from pathlib import Path +from re import Pattern +from types import ModuleType +from typing import ( + TYPE_CHECKING, + Annotated, + Any, + Callable, + ClassVar, + Generic, + Literal, + TypeVar, + Union, + cast, +) +from uuid import UUID + +import annotated_types +from annotated_types import BaseMetadata, MaxLen, MinLen +from pydantic_core import CoreSchema, PydanticCustomError, SchemaSerializer, core_schema +from typing_extensions import Protocol, TypeAlias, TypeAliasType, deprecated, get_args, get_origin + +from ._internal import _fields, _internal_dataclass, _utils, _validators +from ._migration import getattr_migration +from .annotated_handlers import GetCoreSchemaHandler, GetJsonSchemaHandler +from .errors import PydanticUserError +from .json_schema import JsonSchemaValue +from .warnings import PydanticDeprecatedSince20 + +if TYPE_CHECKING: + from ._internal._core_metadata import CoreMetadata + +__all__ = ( + 'Strict', + 'StrictStr', + 'SocketPath', + 'conbytes', + 'conlist', + 'conset', + 'confrozenset', + 'constr', + 'ImportString', + 'conint', + 'PositiveInt', + 'NegativeInt', + 'NonNegativeInt', + 'NonPositiveInt', + 'confloat', + 'PositiveFloat', + 'NegativeFloat', + 'NonNegativeFloat', + 'NonPositiveFloat', + 'FiniteFloat', + 'condecimal', + 'UUID1', + 'UUID3', + 'UUID4', + 'UUID5', + 'UUID6', + 'UUID7', + 'UUID8', + 'FilePath', + 'DirectoryPath', + 'NewPath', + 'Json', + 'Secret', + 'SecretStr', + 'SecretBytes', + 'StrictBool', + 'StrictBytes', + 'StrictInt', + 'StrictFloat', + 'PaymentCardNumber', + 'ByteSize', + 'PastDate', + 'FutureDate', + 'PastDatetime', + 'FutureDatetime', + 'condate', + 'AwareDatetime', + 'NaiveDatetime', + 'AllowInfNan', + 'EncoderProtocol', + 'EncodedBytes', + 'EncodedStr', + 'Base64Encoder', + 'Base64Bytes', + 'Base64Str', + 'Base64UrlBytes', + 'Base64UrlStr', + 'GetPydanticSchema', + 'StringConstraints', + 'Tag', + 'Discriminator', + 'JsonValue', + 'OnErrorOmit', + 'FailFast', +) + + +T = TypeVar('T') + + +@_dataclasses.dataclass +class Strict(_fields.PydanticMetadata, BaseMetadata): + """!!! abstract "Usage Documentation" + [Strict Mode with `Annotated` `Strict`](../concepts/strict_mode.md#strict-mode-with-annotated-strict) + + A field metadata class to indicate that a field should be validated in strict mode. + Use this class as an annotation via [`Annotated`](https://docs.python.org/3/library/typing.html#typing.Annotated), as seen below. + + Attributes: + strict: Whether to validate the field in strict mode. + + Example: + ```python + from typing import Annotated + + from pydantic.types import Strict + + StrictBool = Annotated[bool, Strict()] + ``` + """ + + strict: bool = True + + def __hash__(self) -> int: + return hash(self.strict) + + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ BOOLEAN TYPES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +StrictBool = Annotated[bool, Strict()] +"""A boolean that must be either ``True`` or ``False``.""" + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ INTEGER TYPES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + + +def conint( + *, + strict: bool | None = None, + gt: int | None = None, + ge: int | None = None, + lt: int | None = None, + le: int | None = None, + multiple_of: int | None = None, +) -> type[int]: + """ + !!! warning "Discouraged" + This function is **discouraged** in favor of using + [`Annotated`](https://docs.python.org/3/library/typing.html#typing.Annotated) with + [`Field`][pydantic.fields.Field] instead. + + This function will be **deprecated** in Pydantic 3.0. + + The reason is that `conint` returns a type, which doesn't play well with static analysis tools. + + === ":x: Don't do this" + ```python + from pydantic import BaseModel, conint + + class Foo(BaseModel): + bar: conint(strict=True, gt=0) + ``` + + === ":white_check_mark: Do this" + ```python + from typing import Annotated + + from pydantic import BaseModel, Field + + class Foo(BaseModel): + bar: Annotated[int, Field(strict=True, gt=0)] + ``` + + A wrapper around `int` that allows for additional constraints. + + Args: + strict: Whether to validate the integer in strict mode. Defaults to `None`. + gt: The value must be greater than this. + ge: The value must be greater than or equal to this. + lt: The value must be less than this. + le: The value must be less than or equal to this. + multiple_of: The value must be a multiple of this. + + Returns: + The wrapped integer type. + + ```python + from pydantic import BaseModel, ValidationError, conint + + class ConstrainedExample(BaseModel): + constrained_int: conint(gt=1) + + m = ConstrainedExample(constrained_int=2) + print(repr(m)) + #> ConstrainedExample(constrained_int=2) + + try: + ConstrainedExample(constrained_int=0) + except ValidationError as e: + print(e.errors()) + ''' + [ + { + 'type': 'greater_than', + 'loc': ('constrained_int',), + 'msg': 'Input should be greater than 1', + 'input': 0, + 'ctx': {'gt': 1}, + 'url': 'https://errors.pydantic.dev/2/v/greater_than', + } + ] + ''' + ``` + + """ # noqa: D212 + return Annotated[ # pyright: ignore[reportReturnType] + int, + Strict(strict) if strict is not None else None, + annotated_types.Interval(gt=gt, ge=ge, lt=lt, le=le), + annotated_types.MultipleOf(multiple_of) if multiple_of is not None else None, + ] + + +PositiveInt = Annotated[int, annotated_types.Gt(0)] +"""An integer that must be greater than zero. + +```python +from pydantic import BaseModel, PositiveInt, ValidationError + +class Model(BaseModel): + positive_int: PositiveInt + +m = Model(positive_int=1) +print(repr(m)) +#> Model(positive_int=1) + +try: + Model(positive_int=-1) +except ValidationError as e: + print(e.errors()) + ''' + [ + { + 'type': 'greater_than', + 'loc': ('positive_int',), + 'msg': 'Input should be greater than 0', + 'input': -1, + 'ctx': {'gt': 0}, + 'url': 'https://errors.pydantic.dev/2/v/greater_than', + } + ] + ''' +``` +""" +NegativeInt = Annotated[int, annotated_types.Lt(0)] +"""An integer that must be less than zero. + +```python +from pydantic import BaseModel, NegativeInt, ValidationError + +class Model(BaseModel): + negative_int: NegativeInt + +m = Model(negative_int=-1) +print(repr(m)) +#> Model(negative_int=-1) + +try: + Model(negative_int=1) +except ValidationError as e: + print(e.errors()) + ''' + [ + { + 'type': 'less_than', + 'loc': ('negative_int',), + 'msg': 'Input should be less than 0', + 'input': 1, + 'ctx': {'lt': 0}, + 'url': 'https://errors.pydantic.dev/2/v/less_than', + } + ] + ''' +``` +""" +NonPositiveInt = Annotated[int, annotated_types.Le(0)] +"""An integer that must be less than or equal to zero. + +```python +from pydantic import BaseModel, NonPositiveInt, ValidationError + +class Model(BaseModel): + non_positive_int: NonPositiveInt + +m = Model(non_positive_int=0) +print(repr(m)) +#> Model(non_positive_int=0) + +try: + Model(non_positive_int=1) +except ValidationError as e: + print(e.errors()) + ''' + [ + { + 'type': 'less_than_equal', + 'loc': ('non_positive_int',), + 'msg': 'Input should be less than or equal to 0', + 'input': 1, + 'ctx': {'le': 0}, + 'url': 'https://errors.pydantic.dev/2/v/less_than_equal', + } + ] + ''' +``` +""" +NonNegativeInt = Annotated[int, annotated_types.Ge(0)] +"""An integer that must be greater than or equal to zero. + +```python +from pydantic import BaseModel, NonNegativeInt, ValidationError + +class Model(BaseModel): + non_negative_int: NonNegativeInt + +m = Model(non_negative_int=0) +print(repr(m)) +#> Model(non_negative_int=0) + +try: + Model(non_negative_int=-1) +except ValidationError as e: + print(e.errors()) + ''' + [ + { + 'type': 'greater_than_equal', + 'loc': ('non_negative_int',), + 'msg': 'Input should be greater than or equal to 0', + 'input': -1, + 'ctx': {'ge': 0}, + 'url': 'https://errors.pydantic.dev/2/v/greater_than_equal', + } + ] + ''' +``` +""" +StrictInt = Annotated[int, Strict()] +"""An integer that must be validated in strict mode. + +```python +from pydantic import BaseModel, StrictInt, ValidationError + +class StrictIntModel(BaseModel): + strict_int: StrictInt + +try: + StrictIntModel(strict_int=3.14159) +except ValidationError as e: + print(e) + ''' + 1 validation error for StrictIntModel + strict_int + Input should be a valid integer [type=int_type, input_value=3.14159, input_type=float] + ''' +``` +""" + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ FLOAT TYPES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + + +@_dataclasses.dataclass +class AllowInfNan(_fields.PydanticMetadata): + """A field metadata class to indicate that a field should allow `-inf`, `inf`, and `nan`. + + Use this class as an annotation via [`Annotated`](https://docs.python.org/3/library/typing.html#typing.Annotated), as seen below. + + Attributes: + allow_inf_nan: Whether to allow `-inf`, `inf`, and `nan`. Defaults to `True`. + + Example: + ```python + from typing import Annotated + + from pydantic.types import AllowInfNan + + LaxFloat = Annotated[float, AllowInfNan()] + ``` + """ + + allow_inf_nan: bool = True + + def __hash__(self) -> int: + return hash(self.allow_inf_nan) + + +def confloat( + *, + strict: bool | None = None, + gt: float | None = None, + ge: float | None = None, + lt: float | None = None, + le: float | None = None, + multiple_of: float | None = None, + allow_inf_nan: bool | None = None, +) -> type[float]: + """ + !!! warning "Discouraged" + This function is **discouraged** in favor of using + [`Annotated`](https://docs.python.org/3/library/typing.html#typing.Annotated) with + [`Field`][pydantic.fields.Field] instead. + + This function will be **deprecated** in Pydantic 3.0. + + The reason is that `confloat` returns a type, which doesn't play well with static analysis tools. + + === ":x: Don't do this" + ```python + from pydantic import BaseModel, confloat + + class Foo(BaseModel): + bar: confloat(strict=True, gt=0) + ``` + + === ":white_check_mark: Do this" + ```python + from typing import Annotated + + from pydantic import BaseModel, Field + + class Foo(BaseModel): + bar: Annotated[float, Field(strict=True, gt=0)] + ``` + + A wrapper around `float` that allows for additional constraints. + + Args: + strict: Whether to validate the float in strict mode. + gt: The value must be greater than this. + ge: The value must be greater than or equal to this. + lt: The value must be less than this. + le: The value must be less than or equal to this. + multiple_of: The value must be a multiple of this. + allow_inf_nan: Whether to allow `-inf`, `inf`, and `nan`. + + Returns: + The wrapped float type. + + ```python + from pydantic import BaseModel, ValidationError, confloat + + class ConstrainedExample(BaseModel): + constrained_float: confloat(gt=1.0) + + m = ConstrainedExample(constrained_float=1.1) + print(repr(m)) + #> ConstrainedExample(constrained_float=1.1) + + try: + ConstrainedExample(constrained_float=0.9) + except ValidationError as e: + print(e.errors()) + ''' + [ + { + 'type': 'greater_than', + 'loc': ('constrained_float',), + 'msg': 'Input should be greater than 1', + 'input': 0.9, + 'ctx': {'gt': 1.0}, + 'url': 'https://errors.pydantic.dev/2/v/greater_than', + } + ] + ''' + ``` + """ # noqa: D212 + return Annotated[ # pyright: ignore[reportReturnType] + float, + Strict(strict) if strict is not None else None, + annotated_types.Interval(gt=gt, ge=ge, lt=lt, le=le), + annotated_types.MultipleOf(multiple_of) if multiple_of is not None else None, + AllowInfNan(allow_inf_nan) if allow_inf_nan is not None else None, + ] + + +PositiveFloat = Annotated[float, annotated_types.Gt(0)] +"""A float that must be greater than zero. + +```python +from pydantic import BaseModel, PositiveFloat, ValidationError + +class Model(BaseModel): + positive_float: PositiveFloat + +m = Model(positive_float=1.0) +print(repr(m)) +#> Model(positive_float=1.0) + +try: + Model(positive_float=-1.0) +except ValidationError as e: + print(e.errors()) + ''' + [ + { + 'type': 'greater_than', + 'loc': ('positive_float',), + 'msg': 'Input should be greater than 0', + 'input': -1.0, + 'ctx': {'gt': 0.0}, + 'url': 'https://errors.pydantic.dev/2/v/greater_than', + } + ] + ''' +``` +""" +NegativeFloat = Annotated[float, annotated_types.Lt(0)] +"""A float that must be less than zero. + +```python +from pydantic import BaseModel, NegativeFloat, ValidationError + +class Model(BaseModel): + negative_float: NegativeFloat + +m = Model(negative_float=-1.0) +print(repr(m)) +#> Model(negative_float=-1.0) + +try: + Model(negative_float=1.0) +except ValidationError as e: + print(e.errors()) + ''' + [ + { + 'type': 'less_than', + 'loc': ('negative_float',), + 'msg': 'Input should be less than 0', + 'input': 1.0, + 'ctx': {'lt': 0.0}, + 'url': 'https://errors.pydantic.dev/2/v/less_than', + } + ] + ''' +``` +""" +NonPositiveFloat = Annotated[float, annotated_types.Le(0)] +"""A float that must be less than or equal to zero. + +```python +from pydantic import BaseModel, NonPositiveFloat, ValidationError + +class Model(BaseModel): + non_positive_float: NonPositiveFloat + +m = Model(non_positive_float=0.0) +print(repr(m)) +#> Model(non_positive_float=0.0) + +try: + Model(non_positive_float=1.0) +except ValidationError as e: + print(e.errors()) + ''' + [ + { + 'type': 'less_than_equal', + 'loc': ('non_positive_float',), + 'msg': 'Input should be less than or equal to 0', + 'input': 1.0, + 'ctx': {'le': 0.0}, + 'url': 'https://errors.pydantic.dev/2/v/less_than_equal', + } + ] + ''' +``` +""" +NonNegativeFloat = Annotated[float, annotated_types.Ge(0)] +"""A float that must be greater than or equal to zero. + +```python +from pydantic import BaseModel, NonNegativeFloat, ValidationError + +class Model(BaseModel): + non_negative_float: NonNegativeFloat + +m = Model(non_negative_float=0.0) +print(repr(m)) +#> Model(non_negative_float=0.0) + +try: + Model(non_negative_float=-1.0) +except ValidationError as e: + print(e.errors()) + ''' + [ + { + 'type': 'greater_than_equal', + 'loc': ('non_negative_float',), + 'msg': 'Input should be greater than or equal to 0', + 'input': -1.0, + 'ctx': {'ge': 0.0}, + 'url': 'https://errors.pydantic.dev/2/v/greater_than_equal', + } + ] + ''' +``` +""" +StrictFloat = Annotated[float, Strict(True)] +"""A float that must be validated in strict mode. + +```python +from pydantic import BaseModel, StrictFloat, ValidationError + +class StrictFloatModel(BaseModel): + strict_float: StrictFloat + +try: + StrictFloatModel(strict_float='1.0') +except ValidationError as e: + print(e) + ''' + 1 validation error for StrictFloatModel + strict_float + Input should be a valid number [type=float_type, input_value='1.0', input_type=str] + ''' +``` +""" +FiniteFloat = Annotated[float, AllowInfNan(False)] +"""A float that must be finite (not ``-inf``, ``inf``, or ``nan``). + +```python +from pydantic import BaseModel, FiniteFloat + +class Model(BaseModel): + finite: FiniteFloat + +m = Model(finite=1.0) +print(m) +#> finite=1.0 +``` +""" + + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ BYTES TYPES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + + +def conbytes( + *, + min_length: int | None = None, + max_length: int | None = None, + strict: bool | None = None, +) -> type[bytes]: + """A wrapper around `bytes` that allows for additional constraints. + + Args: + min_length: The minimum length of the bytes. + max_length: The maximum length of the bytes. + strict: Whether to validate the bytes in strict mode. + + Returns: + The wrapped bytes type. + """ + return Annotated[ # pyright: ignore[reportReturnType] + bytes, + Strict(strict) if strict is not None else None, + annotated_types.Len(min_length or 0, max_length), + ] + + +StrictBytes = Annotated[bytes, Strict()] +"""A bytes that must be validated in strict mode.""" + + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ STRING TYPES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + + +@_dataclasses.dataclass(frozen=True) +class StringConstraints(annotated_types.GroupedMetadata): + """!!! abstract "Usage Documentation" + [String types](./standard_library_types.md#strings) + + A field metadata class to apply constraints to `str` types. + Use this class as an annotation via [`Annotated`](https://docs.python.org/3/library/typing.html#typing.Annotated), as seen below. + + Attributes: + strip_whitespace: Whether to remove leading and trailing whitespace. + to_upper: Whether to convert the string to uppercase. + to_lower: Whether to convert the string to lowercase. + strict: Whether to validate the string in strict mode. + min_length: The minimum length of the string. + max_length: The maximum length of the string. + pattern: A regex pattern that the string must match. + ascii_only: Whether the string should contain only ASCII characters. + + Example: + ```python + from typing import Annotated + + from pydantic.types import StringConstraints + + ConstrainedStr = Annotated[str, StringConstraints(min_length=1, max_length=10)] + ``` + """ + + strip_whitespace: bool | None = None + to_upper: bool | None = None + to_lower: bool | None = None + strict: bool | None = None + min_length: int | None = None + max_length: int | None = None + pattern: str | Pattern[str] | None = None + ascii_only: bool | None = None + + def __iter__(self) -> Iterator[BaseMetadata]: + if self.min_length is not None: + yield MinLen(self.min_length) + if self.max_length is not None: + yield MaxLen(self.max_length) + if self.strict is not None: + yield Strict(self.strict) + if ( + self.strip_whitespace is not None + or self.pattern is not None + or self.to_lower is not None + or self.to_upper is not None + or self.ascii_only is not None + ): + yield _fields.pydantic_general_metadata( + strip_whitespace=self.strip_whitespace, + to_upper=self.to_upper, + to_lower=self.to_lower, + pattern=self.pattern, + ascii_only=self.ascii_only, + ) + + +def constr( + *, + strip_whitespace: bool | None = None, + to_upper: bool | None = None, + to_lower: bool | None = None, + strict: bool | None = None, + min_length: int | None = None, + max_length: int | None = None, + pattern: str | Pattern[str] | None = None, + ascii_only: bool | None = None, +) -> type[str]: + """ + !!! warning "Discouraged" + This function is **discouraged** in favor of using + [`Annotated`](https://docs.python.org/3/library/typing.html#typing.Annotated) with + [`StringConstraints`][pydantic.types.StringConstraints] instead. + + This function will be **deprecated** in Pydantic 3.0. + + The reason is that `constr` returns a type, which doesn't play well with static analysis tools. + + === ":x: Don't do this" + ```python + from pydantic import BaseModel, constr + + class Foo(BaseModel): + bar: constr(strip_whitespace=True, to_upper=True, pattern=r'^[A-Z]+$') + ``` + + === ":white_check_mark: Do this" + ```python + from typing import Annotated + + from pydantic import BaseModel, StringConstraints + + class Foo(BaseModel): + bar: Annotated[ + str, + StringConstraints( + strip_whitespace=True, to_upper=True, pattern=r'^[A-Z]+$' + ), + ] + ``` + + A wrapper around `str` that allows for additional constraints. + + ```python + from pydantic import BaseModel, constr + + class Foo(BaseModel): + bar: constr(strip_whitespace=True, to_upper=True) + + foo = Foo(bar=' hello ') + print(foo) + #> bar='HELLO' + ``` + + Args: + strip_whitespace: Whether to remove leading and trailing whitespace. + to_upper: Whether to turn all characters to uppercase. + to_lower: Whether to turn all characters to lowercase. + strict: Whether to validate the string in strict mode. + min_length: The minimum length of the string. + max_length: The maximum length of the string. + pattern: A regex pattern to validate the string against. + ascii_only: Whether the string should contain only ASCII characters. + + Returns: + The wrapped string type. + """ # noqa: D212 + return Annotated[ # pyright: ignore[reportReturnType] + str, + StringConstraints( + strip_whitespace=strip_whitespace, + to_upper=to_upper, + to_lower=to_lower, + strict=strict, + min_length=min_length, + max_length=max_length, + pattern=pattern, + ascii_only=ascii_only, + ), + ] + + +StrictStr = Annotated[str, Strict()] +"""A string that must be validated in strict mode.""" + + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~ COLLECTION TYPES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +HashableItemType = TypeVar('HashableItemType', bound=Hashable) + + +def conset( + item_type: type[HashableItemType], *, min_length: int | None = None, max_length: int | None = None +) -> type[set[HashableItemType]]: + """A wrapper around `typing.Set` that allows for additional constraints. + + Args: + item_type: The type of the items in the set. + min_length: The minimum length of the set. + max_length: The maximum length of the set. + + Returns: + The wrapped set type. + """ + return Annotated[set[item_type], annotated_types.Len(min_length or 0, max_length)] # pyright: ignore[reportReturnType] + + +def confrozenset( + item_type: type[HashableItemType], *, min_length: int | None = None, max_length: int | None = None +) -> type[frozenset[HashableItemType]]: + """A wrapper around `typing.FrozenSet` that allows for additional constraints. + + Args: + item_type: The type of the items in the frozenset. + min_length: The minimum length of the frozenset. + max_length: The maximum length of the frozenset. + + Returns: + The wrapped frozenset type. + """ + return Annotated[frozenset[item_type], annotated_types.Len(min_length or 0, max_length)] # pyright: ignore[reportReturnType] + + +AnyItemType = TypeVar('AnyItemType') + + +def conlist( + item_type: type[AnyItemType], + *, + min_length: int | None = None, + max_length: int | None = None, + unique_items: bool | None = None, +) -> type[list[AnyItemType]]: + """A wrapper around [`list`][] that adds validation. + + Args: + item_type: The type of the items in the list. + min_length: The minimum length of the list. Defaults to None. + max_length: The maximum length of the list. Defaults to None. + unique_items: Whether the items in the list must be unique. Defaults to None. + !!! warning Deprecated + The `unique_items` parameter is deprecated, use `Set` instead. + See [this issue](https://github.com/pydantic/pydantic-core/issues/296) for more details. + + Returns: + The wrapped list type. + """ + if unique_items is not None: + raise PydanticUserError( + ( + '`unique_items` is removed, use `Set` instead' + '(this feature is discussed in https://github.com/pydantic/pydantic-core/issues/296)' + ), + code='removed-kwargs', + ) + return Annotated[list[item_type], annotated_types.Len(min_length or 0, max_length)] # pyright: ignore[reportReturnType] + + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~ IMPORT STRING TYPE ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +AnyType = TypeVar('AnyType') +if TYPE_CHECKING: + ImportString = Annotated[AnyType, ...] +else: + + class ImportString: + """A type that can be used to import a Python object from a string. + + `ImportString` expects a string and loads the Python object importable at that dotted path. + Attributes of modules may be separated from the module by `:` or `.`, e.g. if `'math:cos'` is provided, + the resulting field value would be the function `cos`. If a `.` is used and both an attribute and submodule + are present at the same path, the module will be preferred. + + On model instantiation, pointers will be evaluated and imported. There is + some nuance to this behavior, demonstrated in the examples below. + + ```python + import math + + from pydantic import BaseModel, Field, ImportString, ValidationError + + class ImportThings(BaseModel): + obj: ImportString + + # A string value will cause an automatic import + my_cos = ImportThings(obj='math.cos') + + # You can use the imported function as you would expect + cos_of_0 = my_cos.obj(0) + assert cos_of_0 == 1 + + # A string whose value cannot be imported will raise an error + try: + ImportThings(obj='foo.bar') + except ValidationError as e: + print(e) + ''' + 1 validation error for ImportThings + obj + Invalid python path: No module named 'foo' [type=import_error, input_value='foo.bar', input_type=str] + ''' + + # Actual python objects can be assigned as well + my_cos = ImportThings(obj=math.cos) + my_cos_2 = ImportThings(obj='math.cos') + my_cos_3 = ImportThings(obj='math:cos') + assert my_cos == my_cos_2 == my_cos_3 + + # You can set default field value either as Python object: + class ImportThingsDefaultPyObj(BaseModel): + obj: ImportString = math.cos + + # or as a string value (but only if used with `validate_default=True`) + class ImportThingsDefaultString(BaseModel): + obj: ImportString = Field(default='math.cos', validate_default=True) + + my_cos_default1 = ImportThingsDefaultPyObj() + my_cos_default2 = ImportThingsDefaultString() + assert my_cos_default1.obj == my_cos_default2.obj == math.cos + + # note: this will not work! + class ImportThingsMissingValidateDefault(BaseModel): + obj: ImportString = 'math.cos' + + my_cos_default3 = ImportThingsMissingValidateDefault() + assert my_cos_default3.obj == 'math.cos' # just string, not evaluated + ``` + + Serializing an `ImportString` type to json is also possible. + + ```python + from pydantic import BaseModel, ImportString + + class ImportThings(BaseModel): + obj: ImportString + + # Create an instance + m = ImportThings(obj='math.cos') + print(m) + #> obj= + print(m.model_dump_json()) + #> {"obj":"math.cos"} + ``` + """ + + @classmethod + def __class_getitem__(cls, item: AnyType) -> AnyType: + return Annotated[item, cls()] + + @classmethod + def __get_pydantic_core_schema__( + cls, source: type[Any], handler: GetCoreSchemaHandler + ) -> core_schema.CoreSchema: + serializer = core_schema.plain_serializer_function_ser_schema(cls._serialize, when_used='json') + if cls is source: + # Treat bare usage of ImportString (`schema is None`) as the same as ImportString[Any] + return core_schema.no_info_plain_validator_function( + function=_validators.import_string, serialization=serializer + ) + else: + return core_schema.no_info_before_validator_function( + function=_validators.import_string, schema=handler(source), serialization=serializer + ) + + @classmethod + def __get_pydantic_json_schema__(cls, cs: CoreSchema, handler: GetJsonSchemaHandler) -> JsonSchemaValue: + return handler(core_schema.str_schema()) + + @staticmethod + def _serialize(v: Any) -> str: + if isinstance(v, ModuleType): + return v.__name__ + elif hasattr(v, '__module__') and hasattr(v, '__name__'): + return f'{v.__module__}.{v.__name__}' + # Handle special cases for sys.XXX streams + # if we see more of these, we should consider a more general solution + elif hasattr(v, 'name'): + if v.name == '': + return 'sys.stdout' + elif v.name == '': + return 'sys.stdin' + elif v.name == '': + return 'sys.stderr' + return v + + def __repr__(self) -> str: + return 'ImportString' + + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ DECIMAL TYPES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + + +def condecimal( + *, + strict: bool | None = None, + gt: int | Decimal | None = None, + ge: int | Decimal | None = None, + lt: int | Decimal | None = None, + le: int | Decimal | None = None, + multiple_of: int | Decimal | None = None, + max_digits: int | None = None, + decimal_places: int | None = None, + allow_inf_nan: bool | None = None, +) -> type[Decimal]: + """ + !!! warning "Discouraged" + This function is **discouraged** in favor of using + [`Annotated`](https://docs.python.org/3/library/typing.html#typing.Annotated) with + [`Field`][pydantic.fields.Field] instead. + + This function will be **deprecated** in Pydantic 3.0. + + The reason is that `condecimal` returns a type, which doesn't play well with static analysis tools. + + === ":x: Don't do this" + ```python + from pydantic import BaseModel, condecimal + + class Foo(BaseModel): + bar: condecimal(strict=True, allow_inf_nan=True) + ``` + + === ":white_check_mark: Do this" + ```python + from decimal import Decimal + from typing import Annotated + + from pydantic import BaseModel, Field + + class Foo(BaseModel): + bar: Annotated[Decimal, Field(strict=True, allow_inf_nan=True)] + ``` + + A wrapper around Decimal that adds validation. + + Args: + strict: Whether to validate the value in strict mode. Defaults to `None`. + gt: The value must be greater than this. Defaults to `None`. + ge: The value must be greater than or equal to this. Defaults to `None`. + lt: The value must be less than this. Defaults to `None`. + le: The value must be less than or equal to this. Defaults to `None`. + multiple_of: The value must be a multiple of this. Defaults to `None`. + max_digits: The maximum number of digits. Defaults to `None`. + decimal_places: The number of decimal places. Defaults to `None`. + allow_inf_nan: Whether to allow infinity and NaN. Defaults to `None`. + + ```python + from decimal import Decimal + + from pydantic import BaseModel, ValidationError, condecimal + + class ConstrainedExample(BaseModel): + constrained_decimal: condecimal(gt=Decimal('1.0')) + + m = ConstrainedExample(constrained_decimal=Decimal('1.1')) + print(repr(m)) + #> ConstrainedExample(constrained_decimal=Decimal('1.1')) + + try: + ConstrainedExample(constrained_decimal=Decimal('0.9')) + except ValidationError as e: + print(e.errors()) + ''' + [ + { + 'type': 'greater_than', + 'loc': ('constrained_decimal',), + 'msg': 'Input should be greater than 1.0', + 'input': Decimal('0.9'), + 'ctx': {'gt': Decimal('1.0')}, + 'url': 'https://errors.pydantic.dev/2/v/greater_than', + } + ] + ''' + ``` + """ # noqa: D212 + return Annotated[ # pyright: ignore[reportReturnType] + Decimal, + Strict(strict) if strict is not None else None, + annotated_types.Interval(gt=gt, ge=ge, lt=lt, le=le), + annotated_types.MultipleOf(multiple_of) if multiple_of is not None else None, + _fields.pydantic_general_metadata(max_digits=max_digits, decimal_places=decimal_places), + AllowInfNan(allow_inf_nan) if allow_inf_nan is not None else None, + ] + + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ UUID TYPES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + + +@_dataclasses.dataclass(**_internal_dataclass.slots_true) +class UuidVersion: + """A field metadata class to indicate a [UUID](https://docs.python.org/3/library/uuid.html) version. + + Use this class as an annotation via [`Annotated`](https://docs.python.org/3/library/typing.html#typing.Annotated), as seen below. + + Attributes: + uuid_version: The version of the UUID. Must be one of 1, 3, 4, 5, 6, 7 or 8. + + Example: + ```python + from typing import Annotated + from uuid import UUID + + from pydantic.types import UuidVersion + + UUID1 = Annotated[UUID, UuidVersion(1)] + ``` + """ + + uuid_version: Literal[1, 3, 4, 5, 6, 7, 8] + + def __get_pydantic_json_schema__( + self, core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler + ) -> JsonSchemaValue: + field_schema = handler(core_schema) + field_schema.pop('anyOf', None) # remove the bytes/str union + field_schema.update(type='string', format=f'uuid{self.uuid_version}') + return field_schema + + def __get_pydantic_core_schema__(self, source: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: + schema = handler(source) + _check_annotated_type(schema['type'], 'uuid', self.__class__.__name__) + schema['version'] = self.uuid_version # type: ignore + return schema + + def __hash__(self) -> int: + return hash(self.uuid_version) + + +UUID1 = Annotated[UUID, UuidVersion(1)] +"""A [UUID](https://docs.python.org/3/library/uuid.html) that must be version 1. + +```python +import uuid + +from pydantic import UUID1, BaseModel + +class Model(BaseModel): + uuid1: UUID1 + +Model(uuid1=uuid.uuid1()) +``` +""" +UUID3 = Annotated[UUID, UuidVersion(3)] +"""A [UUID](https://docs.python.org/3/library/uuid.html) that must be version 3. + +```python +import uuid + +from pydantic import UUID3, BaseModel + +class Model(BaseModel): + uuid3: UUID3 + +Model(uuid3=uuid.uuid3(uuid.NAMESPACE_DNS, 'pydantic.org')) +``` +""" +UUID4 = Annotated[UUID, UuidVersion(4)] +"""A [UUID](https://docs.python.org/3/library/uuid.html) that must be version 4. + +```python +import uuid + +from pydantic import UUID4, BaseModel + +class Model(BaseModel): + uuid4: UUID4 + +Model(uuid4=uuid.uuid4()) +``` +""" +UUID5 = Annotated[UUID, UuidVersion(5)] +"""A [UUID](https://docs.python.org/3/library/uuid.html) that must be version 5. + +```python +import uuid + +from pydantic import UUID5, BaseModel + +class Model(BaseModel): + uuid5: UUID5 + +Model(uuid5=uuid.uuid5(uuid.NAMESPACE_DNS, 'pydantic.org')) +``` +""" +UUID6 = Annotated[UUID, UuidVersion(6)] +"""A [UUID](https://docs.python.org/3/library/uuid.html) that must be version 6. + +```python +import uuid + +from pydantic import UUID6, BaseModel + +class Model(BaseModel): + uuid6: UUID6 + +Model(uuid6=uuid.UUID('1efea953-c2d6-6790-aa0a-69db8c87df97')) +``` +""" +UUID7 = Annotated[UUID, UuidVersion(7)] +"""A [UUID](https://docs.python.org/3/library/uuid.html) that must be version 7. + +```python +import uuid + +from pydantic import UUID7, BaseModel + +class Model(BaseModel): + uuid7: UUID7 + +Model(uuid7=uuid.UUID('0194fdcb-1c47-7a09-b52c-561154de0b4a')) +``` +""" +UUID8 = Annotated[UUID, UuidVersion(8)] +"""A [UUID](https://docs.python.org/3/library/uuid.html) that must be version 8. + +```python +import uuid + +from pydantic import UUID8, BaseModel + +class Model(BaseModel): + uuid8: UUID8 + +Model(uuid8=uuid.UUID('81a0b92e-6078-8551-9c81-8ccb666bdab8')) +``` +""" + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ PATH TYPES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + + +@_dataclasses.dataclass +class PathType: + path_type: Literal['file', 'dir', 'new', 'socket'] + + def __get_pydantic_json_schema__( + self, core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler + ) -> JsonSchemaValue: + field_schema = handler(core_schema) + format_conversion = {'file': 'file-path', 'dir': 'directory-path'} + field_schema.update(format=format_conversion.get(self.path_type, 'path'), type='string') + return field_schema + + def __get_pydantic_core_schema__(self, source: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: + function_lookup = { + 'file': cast(core_schema.WithInfoValidatorFunction, self.validate_file), + 'dir': cast(core_schema.WithInfoValidatorFunction, self.validate_directory), + 'new': cast(core_schema.WithInfoValidatorFunction, self.validate_new), + 'socket': cast(core_schema.WithInfoValidatorFunction, self.validate_socket), + } + + return core_schema.with_info_after_validator_function( + function_lookup[self.path_type], + handler(source), + ) + + @staticmethod + def validate_file(path: Path, _: core_schema.ValidationInfo) -> Path: + if path.is_file(): + return path + else: + raise PydanticCustomError('path_not_file', 'Path does not point to a file') + + @staticmethod + def validate_socket(path: Path, _: core_schema.ValidationInfo) -> Path: + if path.is_socket(): + return path + else: + raise PydanticCustomError('path_not_socket', 'Path does not point to a socket') + + @staticmethod + def validate_directory(path: Path, _: core_schema.ValidationInfo) -> Path: + if path.is_dir(): + return path + else: + raise PydanticCustomError('path_not_directory', 'Path does not point to a directory') + + @staticmethod + def validate_new(path: Path, _: core_schema.ValidationInfo) -> Path: + if path.exists(): + raise PydanticCustomError('path_exists', 'Path already exists') + elif not path.parent.exists(): + raise PydanticCustomError('parent_does_not_exist', 'Parent directory does not exist') + else: + return path + + def __hash__(self) -> int: + return hash(self.path_type) + + +FilePath = Annotated[Path, PathType('file')] +"""A path that must point to a file. + +```python +from pathlib import Path + +from pydantic import BaseModel, FilePath, ValidationError + +class Model(BaseModel): + f: FilePath + +path = Path('text.txt') +path.touch() +m = Model(f='text.txt') +print(m.model_dump()) +#> {'f': PosixPath('text.txt')} +path.unlink() + +path = Path('directory') +path.mkdir(exist_ok=True) +try: + Model(f='directory') # directory +except ValidationError as e: + print(e) + ''' + 1 validation error for Model + f + Path does not point to a file [type=path_not_file, input_value='directory', input_type=str] + ''' +path.rmdir() + +try: + Model(f='not-exists-file') +except ValidationError as e: + print(e) + ''' + 1 validation error for Model + f + Path does not point to a file [type=path_not_file, input_value='not-exists-file', input_type=str] + ''' +``` +""" +DirectoryPath = Annotated[Path, PathType('dir')] +"""A path that must point to a directory. + +```python +from pathlib import Path + +from pydantic import BaseModel, DirectoryPath, ValidationError + +class Model(BaseModel): + f: DirectoryPath + +path = Path('directory/') +path.mkdir() +m = Model(f='directory/') +print(m.model_dump()) +#> {'f': PosixPath('directory')} +path.rmdir() + +path = Path('file.txt') +path.touch() +try: + Model(f='file.txt') # file +except ValidationError as e: + print(e) + ''' + 1 validation error for Model + f + Path does not point to a directory [type=path_not_directory, input_value='file.txt', input_type=str] + ''' +path.unlink() + +try: + Model(f='not-exists-directory') +except ValidationError as e: + print(e) + ''' + 1 validation error for Model + f + Path does not point to a directory [type=path_not_directory, input_value='not-exists-directory', input_type=str] + ''' +``` +""" +NewPath = Annotated[Path, PathType('new')] +"""A path for a new file or directory that must not already exist. The parent directory must already exist.""" + +SocketPath = Annotated[Path, PathType('socket')] +"""A path to an existing socket file""" + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ JSON TYPE ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +if TYPE_CHECKING: + # Json[list[str]] will be recognized by type checkers as list[str] + Json = Annotated[AnyType, ...] + +else: + + class Json: + """A special type wrapper which loads JSON before parsing. + + You can use the `Json` data type to make Pydantic first load a raw JSON string before + validating the loaded data into the parametrized type: + + ```python + from typing import Any + + from pydantic import BaseModel, Json, ValidationError + + class AnyJsonModel(BaseModel): + json_obj: Json[Any] + + class ConstrainedJsonModel(BaseModel): + json_obj: Json[list[int]] + + print(AnyJsonModel(json_obj='{"b": 1}')) + #> json_obj={'b': 1} + print(ConstrainedJsonModel(json_obj='[1, 2, 3]')) + #> json_obj=[1, 2, 3] + + try: + ConstrainedJsonModel(json_obj=12) + except ValidationError as e: + print(e) + ''' + 1 validation error for ConstrainedJsonModel + json_obj + JSON input should be string, bytes or bytearray [type=json_type, input_value=12, input_type=int] + ''' + + try: + ConstrainedJsonModel(json_obj='[a, b]') + except ValidationError as e: + print(e) + ''' + 1 validation error for ConstrainedJsonModel + json_obj + Invalid JSON: expected value at line 1 column 2 [type=json_invalid, input_value='[a, b]', input_type=str] + ''' + + try: + ConstrainedJsonModel(json_obj='["a", "b"]') + except ValidationError as e: + print(e) + ''' + 2 validation errors for ConstrainedJsonModel + json_obj.0 + Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='a', input_type=str] + json_obj.1 + Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='b', input_type=str] + ''' + ``` + + When you dump the model using `model_dump` or `model_dump_json`, the dumped value will be the result of validation, + not the original JSON string. However, you can use the argument `round_trip=True` to get the original JSON string back: + + ```python + from pydantic import BaseModel, Json + + class ConstrainedJsonModel(BaseModel): + json_obj: Json[list[int]] + + print(ConstrainedJsonModel(json_obj='[1, 2, 3]').model_dump_json()) + #> {"json_obj":[1,2,3]} + print( + ConstrainedJsonModel(json_obj='[1, 2, 3]').model_dump_json(round_trip=True) + ) + #> {"json_obj":"[1,2,3]"} + ``` + """ + + @classmethod + def __class_getitem__(cls, item: AnyType) -> AnyType: + return Annotated[item, cls()] + + @classmethod + def __get_pydantic_core_schema__(cls, source: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: + if cls is source: + return core_schema.json_schema(None) + else: + return core_schema.json_schema(handler(source)) + + def __repr__(self) -> str: + return 'Json' + + def __hash__(self) -> int: + return hash(type(self)) + + def __eq__(self, other: Any) -> bool: + return type(other) is type(self) + + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ SECRET TYPES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +# The `Secret` class being conceptually immutable, make the type variable covariant: +SecretType = TypeVar('SecretType', covariant=True) + + +class _SecretBase(Generic[SecretType]): + def __init__(self, secret_value: SecretType) -> None: + self._secret_value: SecretType = secret_value + + def get_secret_value(self) -> SecretType: + """Get the secret value. + + Returns: + The secret value. + """ + return self._secret_value + + def __eq__(self, other: Any) -> bool: + return isinstance(other, self.__class__) and self.get_secret_value() == other.get_secret_value() + + def __hash__(self) -> int: + return hash(self.get_secret_value()) + + def __str__(self) -> str: + return str(self._display()) + + def __repr__(self) -> str: + return f'{self.__class__.__name__}({self._display()!r})' + + def _display(self) -> str | bytes: + raise NotImplementedError + + +def _serialize_secret(value: Secret[SecretType], info: core_schema.SerializationInfo) -> str | Secret[SecretType]: + if info.mode == 'json': + return str(value) + else: + return value + + +class Secret(_SecretBase[SecretType]): + """A generic base class used for defining a field with sensitive information that you do not want to be visible in logging or tracebacks. + + You may either directly parametrize `Secret` with a type, or subclass from `Secret` with a parametrized type. The benefit of subclassing + is that you can define a custom `_display` method, which will be used for `repr()` and `str()` methods. The examples below demonstrate both + ways of using `Secret` to create a new secret type. + + 1. Directly parametrizing `Secret` with a type: + + ```python + from pydantic import BaseModel, Secret + + SecretBool = Secret[bool] + + class Model(BaseModel): + secret_bool: SecretBool + + m = Model(secret_bool=True) + print(m.model_dump()) + #> {'secret_bool': Secret('**********')} + + print(m.model_dump_json()) + #> {"secret_bool":"**********"} + + print(m.secret_bool.get_secret_value()) + #> True + ``` + + 2. Subclassing from parametrized `Secret`: + + ```python + from datetime import date + + from pydantic import BaseModel, Secret + + class SecretDate(Secret[date]): + def _display(self) -> str: + return '****/**/**' + + class Model(BaseModel): + secret_date: SecretDate + + m = Model(secret_date=date(2022, 1, 1)) + print(m.model_dump()) + #> {'secret_date': SecretDate('****/**/**')} + + print(m.model_dump_json()) + #> {"secret_date":"****/**/**"} + + print(m.secret_date.get_secret_value()) + #> 2022-01-01 + ``` + + The value returned by the `_display` method will be used for `repr()` and `str()`. + + You can enforce constraints on the underlying type through annotations: + For example: + + ```python + from typing import Annotated + + from pydantic import BaseModel, Field, Secret, ValidationError + + SecretPosInt = Secret[Annotated[int, Field(gt=0, strict=True)]] + + class Model(BaseModel): + sensitive_int: SecretPosInt + + m = Model(sensitive_int=42) + print(m.model_dump()) + #> {'sensitive_int': Secret('**********')} + + try: + m = Model(sensitive_int=-42) # (1)! + except ValidationError as exc_info: + print(exc_info.errors(include_url=False, include_input=False)) + ''' + [ + { + 'type': 'greater_than', + 'loc': ('sensitive_int',), + 'msg': 'Input should be greater than 0', + 'ctx': {'gt': 0}, + } + ] + ''' + + try: + m = Model(sensitive_int='42') # (2)! + except ValidationError as exc_info: + print(exc_info.errors(include_url=False, include_input=False)) + ''' + [ + { + 'type': 'int_type', + 'loc': ('sensitive_int',), + 'msg': 'Input should be a valid integer', + } + ] + ''' + ``` + + 1. The input value is not greater than 0, so it raises a validation error. + 2. The input value is not an integer, so it raises a validation error because the `SecretPosInt` type has strict mode enabled. + """ + + def _display(self) -> str | bytes: + return '**********' if self.get_secret_value() else '' + + @classmethod + def __get_pydantic_core_schema__(cls, source: type[Any], handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: + inner_type = None + # if origin_type is Secret, then cls is a GenericAlias, and we can extract the inner type directly + origin_type = get_origin(source) + if origin_type is not None: + inner_type = get_args(source)[0] + # otherwise, we need to get the inner type from the base class + else: + bases = getattr(cls, '__orig_bases__', getattr(cls, '__bases__', [])) + for base in bases: + if get_origin(base) is Secret: + inner_type = get_args(base)[0] + if bases == [] or inner_type is None: + raise TypeError( + f"Can't get secret type from {cls.__name__}. " + 'Please use Secret[], or subclass from Secret[] instead.' + ) + + inner_schema = handler.generate_schema(inner_type) # type: ignore + + def validate_secret_value(value, handler) -> Secret[SecretType]: + if isinstance(value, Secret): + value = value.get_secret_value() + validated_inner = handler(value) + return cls(validated_inner) + + return core_schema.json_or_python_schema( + python_schema=core_schema.no_info_wrap_validator_function( + validate_secret_value, + inner_schema, + ), + json_schema=core_schema.no_info_after_validator_function(lambda x: cls(x), inner_schema), + serialization=core_schema.plain_serializer_function_ser_schema( + _serialize_secret, + info_arg=True, + when_used='always', + ), + ) + + __pydantic_serializer__ = SchemaSerializer( + core_schema.any_schema( + serialization=core_schema.plain_serializer_function_ser_schema( + _serialize_secret, + info_arg=True, + when_used='always', + ) + ) + ) + + +def _secret_display(value: SecretType) -> str: # type: ignore + return '**********' if value else '' + + +def _serialize_secret_field( + value: _SecretField[SecretType], info: core_schema.SerializationInfo +) -> str | _SecretField[SecretType]: + if info.mode == 'json': + # we want the output to always be string without the `b'` prefix for bytes, + # hence we just use `secret_display` + return _secret_display(value.get_secret_value()) + else: + return value + + +class _SecretField(_SecretBase[SecretType]): + _inner_schema: ClassVar[CoreSchema] + _error_kind: ClassVar[str] + + @classmethod + def __get_pydantic_core_schema__(cls, source: type[Any], handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: + def get_json_schema(_core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler) -> JsonSchemaValue: + json_schema = handler(cls._inner_schema) + _utils.update_not_none( + json_schema, + type='string', + writeOnly=True, + format='password', + ) + return json_schema + + def get_secret_schema(strict: bool) -> CoreSchema: + inner_schema = {**cls._inner_schema, 'strict': strict} + json_schema = core_schema.no_info_after_validator_function( + source, # construct the type + inner_schema, # pyright: ignore[reportArgumentType] + ) + return core_schema.json_or_python_schema( + python_schema=core_schema.union_schema( + [ + core_schema.is_instance_schema(source), + json_schema, + ], + custom_error_type=cls._error_kind, + ), + json_schema=json_schema, + serialization=core_schema.plain_serializer_function_ser_schema( + _serialize_secret_field, + info_arg=True, + when_used='always', + ), + ) + + return core_schema.lax_or_strict_schema( + lax_schema=get_secret_schema(strict=False), + strict_schema=get_secret_schema(strict=True), + metadata={'pydantic_js_functions': [get_json_schema]}, + ) + + __pydantic_serializer__ = SchemaSerializer( + core_schema.any_schema( + serialization=core_schema.plain_serializer_function_ser_schema( + _serialize_secret_field, + info_arg=True, + when_used='always', + ) + ) + ) + + +class SecretStr(_SecretField[str]): + """A string used for storing sensitive information that you do not want to be visible in logging or tracebacks. + + When the secret value is nonempty, it is displayed as `'**********'` instead of the underlying value in + calls to `repr()` and `str()`. If the value _is_ empty, it is displayed as `''`. + + ```python + from pydantic import BaseModel, SecretStr + + class User(BaseModel): + username: str + password: SecretStr + + user = User(username='scolvin', password='password1') + + print(user) + #> username='scolvin' password=SecretStr('**********') + print(user.password.get_secret_value()) + #> password1 + print((SecretStr('password'), SecretStr(''))) + #> (SecretStr('**********'), SecretStr('')) + ``` + + As seen above, by default, [`SecretStr`][pydantic.types.SecretStr] (and [`SecretBytes`][pydantic.types.SecretBytes]) + will be serialized as `**********` when serializing to json. + + You can use the [`field_serializer`][pydantic.functional_serializers.field_serializer] to dump the + secret as plain-text when serializing to json. + + ```python + from pydantic import BaseModel, SecretBytes, SecretStr, field_serializer + + class Model(BaseModel): + password: SecretStr + password_bytes: SecretBytes + + @field_serializer('password', 'password_bytes', when_used='json') + def dump_secret(self, v): + return v.get_secret_value() + + model = Model(password='IAmSensitive', password_bytes=b'IAmSensitiveBytes') + print(model) + #> password=SecretStr('**********') password_bytes=SecretBytes(b'**********') + print(model.password) + #> ********** + print(model.model_dump()) + ''' + { + 'password': SecretStr('**********'), + 'password_bytes': SecretBytes(b'**********'), + } + ''' + print(model.model_dump_json()) + #> {"password":"IAmSensitive","password_bytes":"IAmSensitiveBytes"} + ``` + """ + + _inner_schema: ClassVar[CoreSchema] = core_schema.str_schema() + _error_kind: ClassVar[str] = 'string_type' + + def __len__(self) -> int: + return len(self._secret_value) + + def _display(self) -> str: + return _secret_display(self._secret_value) + + +class SecretBytes(_SecretField[bytes]): + """A bytes used for storing sensitive information that you do not want to be visible in logging or tracebacks. + + It displays `b'**********'` instead of the string value on `repr()` and `str()` calls. + When the secret value is nonempty, it is displayed as `b'**********'` instead of the underlying value in + calls to `repr()` and `str()`. If the value _is_ empty, it is displayed as `b''`. + + ```python + from pydantic import BaseModel, SecretBytes + + class User(BaseModel): + username: str + password: SecretBytes + + user = User(username='scolvin', password=b'password1') + #> username='scolvin' password=SecretBytes(b'**********') + print(user.password.get_secret_value()) + #> b'password1' + print((SecretBytes(b'password'), SecretBytes(b''))) + #> (SecretBytes(b'**********'), SecretBytes(b'')) + ``` + """ + + _inner_schema: ClassVar[CoreSchema] = core_schema.bytes_schema() + _error_kind: ClassVar[str] = 'bytes_type' + + def __len__(self) -> int: + return len(self._secret_value) + + def _display(self) -> bytes: + return _secret_display(self._secret_value).encode() + + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ PAYMENT CARD TYPES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + + +class PaymentCardBrand(str, Enum): + amex = 'American Express' + mastercard = 'Mastercard' + visa = 'Visa' + other = 'other' + + def __str__(self) -> str: + return self.value + + +@deprecated( + 'The `PaymentCardNumber` class is deprecated, use `pydantic_extra_types` instead. ' + 'See https://docs.pydantic.dev/latest/api/pydantic_extra_types_payment/#pydantic_extra_types.payment.PaymentCardNumber.', + category=PydanticDeprecatedSince20, +) +class PaymentCardNumber(str): + """Based on: https://en.wikipedia.org/wiki/Payment_card_number.""" + + strip_whitespace: ClassVar[bool] = True + min_length: ClassVar[int] = 12 + max_length: ClassVar[int] = 19 + bin: str + last4: str + brand: PaymentCardBrand + + def __init__(self, card_number: str): + self.validate_digits(card_number) + + card_number = self.validate_luhn_check_digit(card_number) + + self.bin = card_number[:6] + self.last4 = card_number[-4:] + self.brand = self.validate_brand(card_number) + + @classmethod + def __get_pydantic_core_schema__(cls, source: type[Any], handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: + return core_schema.with_info_after_validator_function( + cls.validate, + core_schema.str_schema( + min_length=cls.min_length, max_length=cls.max_length, strip_whitespace=cls.strip_whitespace + ), + ) + + @classmethod + def validate(cls, input_value: str, /, _: core_schema.ValidationInfo) -> PaymentCardNumber: + """Validate the card number and return a `PaymentCardNumber` instance.""" + return cls(input_value) + + @property + def masked(self) -> str: + """Mask all but the last 4 digits of the card number. + + Returns: + A masked card number string. + """ + num_masked = len(self) - 10 # len(bin) + len(last4) == 10 + return f'{self.bin}{"*" * num_masked}{self.last4}' + + @classmethod + def validate_digits(cls, card_number: str) -> None: + """Validate that the card number is all digits.""" + if not card_number.isdigit(): + raise PydanticCustomError('payment_card_number_digits', 'Card number is not all digits') + + @classmethod + def validate_luhn_check_digit(cls, card_number: str) -> str: + """Based on: https://en.wikipedia.org/wiki/Luhn_algorithm.""" + sum_ = int(card_number[-1]) + length = len(card_number) + parity = length % 2 + for i in range(length - 1): + digit = int(card_number[i]) + if i % 2 == parity: + digit *= 2 + if digit > 9: + digit -= 9 + sum_ += digit + valid = sum_ % 10 == 0 + if not valid: + raise PydanticCustomError('payment_card_number_luhn', 'Card number is not luhn valid') + return card_number + + @staticmethod + def validate_brand(card_number: str) -> PaymentCardBrand: + """Validate length based on BIN for major brands: + https://en.wikipedia.org/wiki/Payment_card_number#Issuer_identification_number_(IIN). + """ + if card_number[0] == '4': + brand = PaymentCardBrand.visa + elif 51 <= int(card_number[:2]) <= 55: + brand = PaymentCardBrand.mastercard + elif card_number[:2] in {'34', '37'}: + brand = PaymentCardBrand.amex + else: + brand = PaymentCardBrand.other + + required_length: None | int | str = None + if brand in PaymentCardBrand.mastercard: + required_length = 16 + valid = len(card_number) == required_length + elif brand == PaymentCardBrand.visa: + required_length = '13, 16 or 19' + valid = len(card_number) in {13, 16, 19} + elif brand == PaymentCardBrand.amex: + required_length = 15 + valid = len(card_number) == required_length + else: + valid = True + + if not valid: + raise PydanticCustomError( + 'payment_card_number_brand', + 'Length for a {brand} card must be {required_length}', + {'brand': brand, 'required_length': required_length}, + ) + return brand + + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ BYTE SIZE TYPE ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + + +class ByteSize(int): + """Converts a string representing a number of bytes with units (such as `'1KB'` or `'11.5MiB'`) into an integer. + + You can use the `ByteSize` data type to (case-insensitively) convert a string representation of a number of bytes into + an integer, and also to print out human-readable strings representing a number of bytes. + + In conformance with [IEC 80000-13 Standard](https://en.wikipedia.org/wiki/ISO/IEC_80000) we interpret `'1KB'` to mean 1000 bytes, + and `'1KiB'` to mean 1024 bytes. In general, including a middle `'i'` will cause the unit to be interpreted as a power of 2, + rather than a power of 10 (so, for example, `'1 MB'` is treated as `1_000_000` bytes, whereas `'1 MiB'` is treated as `1_048_576` bytes). + + !!! info + Note that `1b` will be parsed as "1 byte" and not "1 bit". + + ```python + from pydantic import BaseModel, ByteSize + + class MyModel(BaseModel): + size: ByteSize + + print(MyModel(size=52000).size) + #> 52000 + print(MyModel(size='3000 KiB').size) + #> 3072000 + + m = MyModel(size='50 PB') + print(m.size.human_readable()) + #> 44.4PiB + print(m.size.human_readable(decimal=True)) + #> 50.0PB + print(m.size.human_readable(separator=' ')) + #> 44.4 PiB + + print(m.size.to('TiB')) + #> 45474.73508864641 + ``` + """ + + byte_sizes = { + 'b': 1, + 'kb': 10**3, + 'mb': 10**6, + 'gb': 10**9, + 'tb': 10**12, + 'pb': 10**15, + 'eb': 10**18, + 'kib': 2**10, + 'mib': 2**20, + 'gib': 2**30, + 'tib': 2**40, + 'pib': 2**50, + 'eib': 2**60, + 'bit': 1 / 8, + 'kbit': 10**3 / 8, + 'mbit': 10**6 / 8, + 'gbit': 10**9 / 8, + 'tbit': 10**12 / 8, + 'pbit': 10**15 / 8, + 'ebit': 10**18 / 8, + 'kibit': 2**10 / 8, + 'mibit': 2**20 / 8, + 'gibit': 2**30 / 8, + 'tibit': 2**40 / 8, + 'pibit': 2**50 / 8, + 'eibit': 2**60 / 8, + } + byte_sizes.update({k.lower()[0]: v for k, v in byte_sizes.items() if 'i' not in k}) + + byte_string_pattern = r'^\s*(\d*\.?\d+)\s*(\w+)?' + byte_string_re = re.compile(byte_string_pattern, re.IGNORECASE) + + @classmethod + def __get_pydantic_core_schema__(cls, source: type[Any], handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: + return core_schema.with_info_after_validator_function( + function=cls._validate, + schema=core_schema.union_schema( + [ + core_schema.str_schema(pattern=cls.byte_string_pattern), + core_schema.int_schema(ge=0), + ], + custom_error_type='byte_size', + custom_error_message='could not parse value and unit from byte string', + ), + serialization=core_schema.plain_serializer_function_ser_schema( + int, return_schema=core_schema.int_schema(ge=0) + ), + ) + + @classmethod + def _validate(cls, input_value: Any, /, _: core_schema.ValidationInfo) -> ByteSize: + try: + return cls(int(input_value)) + except ValueError: + pass + + str_match = cls.byte_string_re.match(str(input_value)) + if str_match is None: + raise PydanticCustomError('byte_size', 'could not parse value and unit from byte string') + + scalar, unit = str_match.groups() + if unit is None: + unit = 'b' + + try: + unit_mult = cls.byte_sizes[unit.lower()] + except KeyError: + raise PydanticCustomError('byte_size_unit', 'could not interpret byte unit: {unit}', {'unit': unit}) + + return cls(int(float(scalar) * unit_mult)) + + def human_readable(self, decimal: bool = False, separator: str = '') -> str: + """Converts a byte size to a human readable string. + + Args: + decimal: If True, use decimal units (e.g. 1000 bytes per KB). If False, use binary units + (e.g. 1024 bytes per KiB). + separator: A string used to split the value and unit. Defaults to an empty string (''). + + Returns: + A human readable string representation of the byte size. + """ + if decimal: + divisor = 1000 + units = 'B', 'KB', 'MB', 'GB', 'TB', 'PB' + final_unit = 'EB' + else: + divisor = 1024 + units = 'B', 'KiB', 'MiB', 'GiB', 'TiB', 'PiB' + final_unit = 'EiB' + + num = float(self) + for unit in units: + if abs(num) < divisor: + if unit == 'B': + return f'{num:0.0f}{separator}{unit}' + else: + return f'{num:0.1f}{separator}{unit}' + num /= divisor + + return f'{num:0.1f}{separator}{final_unit}' + + def to(self, unit: str) -> float: + """Converts a byte size to another unit, including both byte and bit units. + + Args: + unit: The unit to convert to. Must be one of the following: B, KB, MB, GB, TB, PB, EB, + KiB, MiB, GiB, TiB, PiB, EiB (byte units) and + bit, kbit, mbit, gbit, tbit, pbit, ebit, + kibit, mibit, gibit, tibit, pibit, eibit (bit units). + + Returns: + The byte size in the new unit. + """ + try: + unit_div = self.byte_sizes[unit.lower()] + except KeyError: + raise PydanticCustomError('byte_size_unit', 'Could not interpret byte unit: {unit}', {'unit': unit}) + + return self / unit_div + + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ DATE TYPES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + + +def _check_annotated_type(annotated_type: str, expected_type: str, annotation: str) -> None: + if annotated_type != expected_type: + raise PydanticUserError(f"'{annotation}' cannot annotate '{annotated_type}'.", code='invalid-annotated-type') + + +if TYPE_CHECKING: + PastDate = Annotated[date, ...] + FutureDate = Annotated[date, ...] +else: + + class PastDate: + """A date in the past.""" + + @classmethod + def __get_pydantic_core_schema__( + cls, source: type[Any], handler: GetCoreSchemaHandler + ) -> core_schema.CoreSchema: + if cls is source: + # used directly as a type + return core_schema.date_schema(now_op='past') + else: + schema = handler(source) + _check_annotated_type(schema['type'], 'date', cls.__name__) + schema['now_op'] = 'past' + return schema + + def __repr__(self) -> str: + return 'PastDate' + + class FutureDate: + """A date in the future.""" + + @classmethod + def __get_pydantic_core_schema__( + cls, source: type[Any], handler: GetCoreSchemaHandler + ) -> core_schema.CoreSchema: + if cls is source: + # used directly as a type + return core_schema.date_schema(now_op='future') + else: + schema = handler(source) + _check_annotated_type(schema['type'], 'date', cls.__name__) + schema['now_op'] = 'future' + return schema + + def __repr__(self) -> str: + return 'FutureDate' + + +def condate( + *, + strict: bool | None = None, + gt: date | None = None, + ge: date | None = None, + lt: date | None = None, + le: date | None = None, +) -> type[date]: + """A wrapper for date that adds constraints. + + Args: + strict: Whether to validate the date value in strict mode. Defaults to `None`. + gt: The value must be greater than this. Defaults to `None`. + ge: The value must be greater than or equal to this. Defaults to `None`. + lt: The value must be less than this. Defaults to `None`. + le: The value must be less than or equal to this. Defaults to `None`. + + Returns: + A date type with the specified constraints. + """ + return Annotated[ # pyright: ignore[reportReturnType] + date, + Strict(strict) if strict is not None else None, + annotated_types.Interval(gt=gt, ge=ge, lt=lt, le=le), + ] + + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ DATETIME TYPES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +if TYPE_CHECKING: + AwareDatetime = Annotated[datetime, ...] + NaiveDatetime = Annotated[datetime, ...] + PastDatetime = Annotated[datetime, ...] + FutureDatetime = Annotated[datetime, ...] + +else: + + class AwareDatetime: + """A datetime that requires timezone info.""" + + @classmethod + def __get_pydantic_core_schema__( + cls, source: type[Any], handler: GetCoreSchemaHandler + ) -> core_schema.CoreSchema: + if cls is source: + # used directly as a type + return core_schema.datetime_schema(tz_constraint='aware') + else: + schema = handler(source) + _check_annotated_type(schema['type'], 'datetime', cls.__name__) + schema['tz_constraint'] = 'aware' + return schema + + def __repr__(self) -> str: + return 'AwareDatetime' + + class NaiveDatetime: + """A datetime that doesn't require timezone info.""" + + @classmethod + def __get_pydantic_core_schema__( + cls, source: type[Any], handler: GetCoreSchemaHandler + ) -> core_schema.CoreSchema: + if cls is source: + # used directly as a type + return core_schema.datetime_schema(tz_constraint='naive') + else: + schema = handler(source) + _check_annotated_type(schema['type'], 'datetime', cls.__name__) + schema['tz_constraint'] = 'naive' + return schema + + def __repr__(self) -> str: + return 'NaiveDatetime' + + class PastDatetime: + """A datetime that must be in the past.""" + + @classmethod + def __get_pydantic_core_schema__( + cls, source: type[Any], handler: GetCoreSchemaHandler + ) -> core_schema.CoreSchema: + if cls is source: + # used directly as a type + return core_schema.datetime_schema(now_op='past') + else: + schema = handler(source) + _check_annotated_type(schema['type'], 'datetime', cls.__name__) + schema['now_op'] = 'past' + return schema + + def __repr__(self) -> str: + return 'PastDatetime' + + class FutureDatetime: + """A datetime that must be in the future.""" + + @classmethod + def __get_pydantic_core_schema__( + cls, source: type[Any], handler: GetCoreSchemaHandler + ) -> core_schema.CoreSchema: + if cls is source: + # used directly as a type + return core_schema.datetime_schema(now_op='future') + else: + schema = handler(source) + _check_annotated_type(schema['type'], 'datetime', cls.__name__) + schema['now_op'] = 'future' + return schema + + def __repr__(self) -> str: + return 'FutureDatetime' + + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Encoded TYPES ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + + +class EncoderProtocol(Protocol): + """Protocol for encoding and decoding data to and from bytes.""" + + @classmethod + def decode(cls, data: bytes) -> bytes: + """Decode the data using the encoder. + + Args: + data: The data to decode. + + Returns: + The decoded data. + """ + ... + + @classmethod + def encode(cls, value: bytes) -> bytes: + """Encode the data using the encoder. + + Args: + value: The data to encode. + + Returns: + The encoded data. + """ + ... + + @classmethod + def get_json_format(cls) -> str: + """Get the JSON format for the encoded data. + + Returns: + The JSON format for the encoded data. + """ + ... + + +class Base64Encoder(EncoderProtocol): + """Standard (non-URL-safe) Base64 encoder.""" + + @classmethod + def decode(cls, data: bytes) -> bytes: + """Decode the data from base64 encoded bytes to original bytes data. + + Args: + data: The data to decode. + + Returns: + The decoded data. + """ + try: + return base64.b64decode(data) + except ValueError as e: + raise PydanticCustomError('base64_decode', "Base64 decoding error: '{error}'", {'error': str(e)}) + + @classmethod + def encode(cls, value: bytes) -> bytes: + """Encode the data from bytes to a base64 encoded bytes. + + Args: + value: The data to encode. + + Returns: + The encoded data. + """ + return base64.b64encode(value) + + @classmethod + def get_json_format(cls) -> Literal['base64']: + """Get the JSON format for the encoded data. + + Returns: + The JSON format for the encoded data. + """ + return 'base64' + + +class Base64UrlEncoder(EncoderProtocol): + """URL-safe Base64 encoder.""" + + @classmethod + def decode(cls, data: bytes) -> bytes: + """Decode the data from base64 encoded bytes to original bytes data. + + Args: + data: The data to decode. + + Returns: + The decoded data. + """ + try: + return base64.urlsafe_b64decode(data) + except ValueError as e: + raise PydanticCustomError('base64_decode', "Base64 decoding error: '{error}'", {'error': str(e)}) + + @classmethod + def encode(cls, value: bytes) -> bytes: + """Encode the data from bytes to a base64 encoded bytes. + + Args: + value: The data to encode. + + Returns: + The encoded data. + """ + return base64.urlsafe_b64encode(value) + + @classmethod + def get_json_format(cls) -> Literal['base64url']: + """Get the JSON format for the encoded data. + + Returns: + The JSON format for the encoded data. + """ + return 'base64url' + + +@_dataclasses.dataclass(**_internal_dataclass.slots_true) +class EncodedBytes: + """A bytes type that is encoded and decoded using the specified encoder. + + `EncodedBytes` needs an encoder that implements `EncoderProtocol` to operate. + + ```python + from typing import Annotated + + from pydantic import BaseModel, EncodedBytes, EncoderProtocol, ValidationError + + class MyEncoder(EncoderProtocol): + @classmethod + def decode(cls, data: bytes) -> bytes: + if data == b'**undecodable**': + raise ValueError('Cannot decode data') + return data[13:] + + @classmethod + def encode(cls, value: bytes) -> bytes: + return b'**encoded**: ' + value + + @classmethod + def get_json_format(cls) -> str: + return 'my-encoder' + + MyEncodedBytes = Annotated[bytes, EncodedBytes(encoder=MyEncoder)] + + class Model(BaseModel): + my_encoded_bytes: MyEncodedBytes + + # Initialize the model with encoded data + m = Model(my_encoded_bytes=b'**encoded**: some bytes') + + # Access decoded value + print(m.my_encoded_bytes) + #> b'some bytes' + + # Serialize into the encoded form + print(m.model_dump()) + #> {'my_encoded_bytes': b'**encoded**: some bytes'} + + # Validate encoded data + try: + Model(my_encoded_bytes=b'**undecodable**') + except ValidationError as e: + print(e) + ''' + 1 validation error for Model + my_encoded_bytes + Value error, Cannot decode data [type=value_error, input_value=b'**undecodable**', input_type=bytes] + ''' + ``` + """ + + encoder: type[EncoderProtocol] + + def __get_pydantic_json_schema__( + self, core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler + ) -> JsonSchemaValue: + field_schema = handler(core_schema) + field_schema.update(type='string', format=self.encoder.get_json_format()) + return field_schema + + def __get_pydantic_core_schema__(self, source: type[Any], handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: + schema = handler(source) + _check_annotated_type(schema['type'], 'bytes', self.__class__.__name__) + return core_schema.with_info_after_validator_function( + function=self.decode, + schema=schema, + serialization=core_schema.plain_serializer_function_ser_schema(function=self.encode), + ) + + def decode(self, data: bytes, _: core_schema.ValidationInfo) -> bytes: + """Decode the data using the specified encoder. + + Args: + data: The data to decode. + + Returns: + The decoded data. + """ + return self.encoder.decode(data) + + def encode(self, value: bytes) -> bytes: + """Encode the data using the specified encoder. + + Args: + value: The data to encode. + + Returns: + The encoded data. + """ + return self.encoder.encode(value) + + def __hash__(self) -> int: + return hash(self.encoder) + + +@_dataclasses.dataclass(**_internal_dataclass.slots_true) +class EncodedStr: + """A str type that is encoded and decoded using the specified encoder. + + `EncodedStr` needs an encoder that implements `EncoderProtocol` to operate. + + ```python + from typing import Annotated + + from pydantic import BaseModel, EncodedStr, EncoderProtocol, ValidationError + + class MyEncoder(EncoderProtocol): + @classmethod + def decode(cls, data: bytes) -> bytes: + if data == b'**undecodable**': + raise ValueError('Cannot decode data') + return data[13:] + + @classmethod + def encode(cls, value: bytes) -> bytes: + return b'**encoded**: ' + value + + @classmethod + def get_json_format(cls) -> str: + return 'my-encoder' + + MyEncodedStr = Annotated[str, EncodedStr(encoder=MyEncoder)] + + class Model(BaseModel): + my_encoded_str: MyEncodedStr + + # Initialize the model with encoded data + m = Model(my_encoded_str='**encoded**: some str') + + # Access decoded value + print(m.my_encoded_str) + #> some str + + # Serialize into the encoded form + print(m.model_dump()) + #> {'my_encoded_str': '**encoded**: some str'} + + # Validate encoded data + try: + Model(my_encoded_str='**undecodable**') + except ValidationError as e: + print(e) + ''' + 1 validation error for Model + my_encoded_str + Value error, Cannot decode data [type=value_error, input_value='**undecodable**', input_type=str] + ''' + ``` + """ + + encoder: type[EncoderProtocol] + + def __get_pydantic_json_schema__( + self, core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler + ) -> JsonSchemaValue: + field_schema = handler(core_schema) + field_schema.update(type='string', format=self.encoder.get_json_format()) + return field_schema + + def __get_pydantic_core_schema__(self, source: type[Any], handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: + schema = handler(source) + _check_annotated_type(schema['type'], 'str', self.__class__.__name__) + return core_schema.with_info_after_validator_function( + function=self.decode_str, + schema=schema, + serialization=core_schema.plain_serializer_function_ser_schema(function=self.encode_str), + ) + + def decode_str(self, data: str, _: core_schema.ValidationInfo) -> str: + """Decode the data using the specified encoder. + + Args: + data: The data to decode. + + Returns: + The decoded data. + """ + return self.encoder.decode(data.encode()).decode() + + def encode_str(self, value: str) -> str: + """Encode the data using the specified encoder. + + Args: + value: The data to encode. + + Returns: + The encoded data. + """ + return self.encoder.encode(value.encode()).decode() # noqa: UP008 + + def __hash__(self) -> int: + return hash(self.encoder) + + +Base64Bytes = Annotated[bytes, EncodedBytes(encoder=Base64Encoder)] +"""A bytes type that is encoded and decoded using the standard (non-URL-safe) base64 encoder. + +Note: + Under the hood, `Base64Bytes` uses the standard library [`base64.b64encode()`][base64.b64encode] and [`base64.b64decode()`][base64.b64decode] functions. + + As a result, attempting to decode url-safe base64 data using the `Base64Bytes` type may fail or produce an incorrect + decoding. + +/// version-changed | v2.10 +`Base64Bytes` now uses [`base64.b64encode()`][base64.b64encode] and [`base64.b64decode()`][base64.b64decode] +instead of [`base64.encodebytes()`][base64.encodebytes] and [`base64.decodebytes()`][base64.decodebytes]. + +These methods are considered legacy implementation. If you'd still like to use these legacy encoders/decoders, +you can achieve this by creating a custom annotated type, like follows: +```python +import base64 +from typing import Annotated, Literal + +from pydantic_core import PydanticCustomError + +from pydantic import EncodedBytes, EncoderProtocol + +class LegacyBase64Encoder(EncoderProtocol): + @classmethod + def decode(cls, data: bytes) -> bytes: + try: + return base64.decodebytes(data) + except ValueError as e: + raise PydanticCustomError( + 'base64_decode', + "Base64 decoding error: '{error}'", + {'error': str(e)}, + ) + + @classmethod + def encode(cls, value: bytes) -> bytes: + return base64.encodebytes(value) + + @classmethod + def get_json_format(cls) -> Literal['base64']: + return 'base64' + +LegacyBase64Bytes = Annotated[bytes, EncodedBytes(encoder=LegacyBase64Encoder)] +``` +/// + +```python +from pydantic import Base64Bytes, BaseModel, ValidationError + +class Model(BaseModel): + base64_bytes: Base64Bytes + +# Initialize the model with base64 data +m = Model(base64_bytes=b'VGhpcyBpcyB0aGUgd2F5') + +# Access decoded value +print(m.base64_bytes) +#> b'This is the way' + +# Serialize into the base64 form +print(m.model_dump()) +#> {'base64_bytes': b'VGhpcyBpcyB0aGUgd2F5'} + +# Validate base64 data +try: + print(Model(base64_bytes=b'undecodable').base64_bytes) +except ValidationError as e: + print(e) + ''' + 1 validation error for Model + base64_bytes + Base64 decoding error: 'Incorrect padding' [type=base64_decode, input_value=b'undecodable', input_type=bytes] + ''' +``` +""" +Base64Str = Annotated[str, EncodedStr(encoder=Base64Encoder)] +"""A string type that is encoded and decoded using the standard (non-URL-safe) base64 encoder. + +Note: + Under the hood, `Base64Str` uses the standard library [`base64.b64encode()`][base64.b64encode] and [`base64.b64decode()`][base64.b64decode] functions. + + As a result, attempting to decode url-safe base64 data using the `Base64Str` type may fail or produce an incorrect + decoding. + +/// version-changed | v2.10 +`Base64Str` now uses [`base64.b64encode()`][base64.b64encode] and [`base64.b64decode()`][base64.b64decode] +instead of [`base64.encodebytes()`][base64.encodebytes] and [`base64.decodebytes()`][base64.decodebytes]. + +These methods are considered legacy implementation. See the documentation about the [`Base64Bytes`][pydantic.types.Base64Bytes] type +for more information on how to replicate the old behavior with the legacy encoders/decoders. +/// + + +```python +from pydantic import Base64Str, BaseModel, ValidationError + +class Model(BaseModel): + base64_str: Base64Str + +# Initialize the model with base64 data +m = Model(base64_str='VGhlc2UgYXJlbid0IHRoZSBkcm9pZHMgeW91J3JlIGxvb2tpbmcgZm9y') + +# Access decoded value +print(m.base64_str) +#> These aren't the droids you're looking for + +# Serialize into the base64 form +print(m.model_dump()) +#> {'base64_str': 'VGhlc2UgYXJlbid0IHRoZSBkcm9pZHMgeW91J3JlIGxvb2tpbmcgZm9y'} + +# Validate base64 data +try: + print(Model(base64_str='undecodable').base64_str) +except ValidationError as e: + print(e) + ''' + 1 validation error for Model + base64_str + Base64 decoding error: 'Incorrect padding' [type=base64_decode, input_value='undecodable', input_type=str] + ''' +``` +""" +Base64UrlBytes = Annotated[bytes, EncodedBytes(encoder=Base64UrlEncoder)] +"""A bytes type that is encoded and decoded using the URL-safe base64 encoder. + +Note: + Under the hood, `Base64UrlBytes` use standard library `base64.urlsafe_b64encode` and `base64.urlsafe_b64decode` + functions. + + As a result, the `Base64UrlBytes` type can be used to faithfully decode "vanilla" base64 data + (using `'+'` and `'/'`). + +```python +from pydantic import Base64UrlBytes, BaseModel + +class Model(BaseModel): + base64url_bytes: Base64UrlBytes + +# Initialize the model with base64 data +m = Model(base64url_bytes=b'SHc_dHc-TXc==') +print(m) +#> base64url_bytes=b'Hw?tw>Mw' +``` +""" +Base64UrlStr = Annotated[str, EncodedStr(encoder=Base64UrlEncoder)] +"""A str type that is encoded and decoded using the URL-safe base64 encoder. + +Note: + Under the hood, `Base64UrlStr` use standard library `base64.urlsafe_b64encode` and `base64.urlsafe_b64decode` + functions. + + As a result, the `Base64UrlStr` type can be used to faithfully decode "vanilla" base64 data (using `'+'` and `'/'`). + +```python +from pydantic import Base64UrlStr, BaseModel + +class Model(BaseModel): + base64url_str: Base64UrlStr + +# Initialize the model with base64 data +m = Model(base64url_str='SHc_dHc-TXc==') +print(m) +#> base64url_str='Hw?tw>Mw' +``` +""" + + +__getattr__ = getattr_migration(__name__) + + +@_dataclasses.dataclass(**_internal_dataclass.slots_true) +class GetPydanticSchema: + """!!! abstract "Usage Documentation" + [Using `GetPydanticSchema` to Reduce Boilerplate](../concepts/types.md#using-getpydanticschema-to-reduce-boilerplate) + + A convenience class for creating an annotation that provides pydantic custom type hooks. + + This class is intended to eliminate the need to create a custom "marker" which defines the + `__get_pydantic_core_schema__` and `__get_pydantic_json_schema__` custom hook methods. + + For example, to have a field treated by type checkers as `int`, but by pydantic as `Any`, you can do: + ```python + from typing import Annotated, Any + + from pydantic import BaseModel, GetPydanticSchema + + HandleAsAny = GetPydanticSchema(lambda _s, h: h(Any)) + + class Model(BaseModel): + x: Annotated[int, HandleAsAny] # pydantic sees `x: Any` + + print(repr(Model(x='abc').x)) + #> 'abc' + ``` + """ + + get_pydantic_core_schema: Callable[[Any, GetCoreSchemaHandler], CoreSchema] | None = None + get_pydantic_json_schema: Callable[[Any, GetJsonSchemaHandler], JsonSchemaValue] | None = None + + # Note: we may want to consider adding a convenience staticmethod `def for_type(type_: Any) -> GetPydanticSchema:` + # which returns `GetPydanticSchema(lambda _s, h: h(type_))` + + if not TYPE_CHECKING: + # We put `__getattr__` in a non-TYPE_CHECKING block because otherwise, mypy allows arbitrary attribute access + + def __getattr__(self, item: str) -> Any: + """Use this rather than defining `__get_pydantic_core_schema__` etc. to reduce the number of nested calls.""" + if item == '__get_pydantic_core_schema__' and self.get_pydantic_core_schema: + return self.get_pydantic_core_schema + elif item == '__get_pydantic_json_schema__' and self.get_pydantic_json_schema: + return self.get_pydantic_json_schema + else: + return object.__getattribute__(self, item) + + __hash__ = object.__hash__ + + +@_dataclasses.dataclass(**_internal_dataclass.slots_true, frozen=True) +class Tag: + """Provides a way to specify the expected tag to use for a case of a (callable) discriminated union. + + Also provides a way to label a union case in error messages. + + When using a callable `Discriminator`, attach a `Tag` to each case in the `Union` to specify the tag that + should be used to identify that case. For example, in the below example, the `Tag` is used to specify that + if `get_discriminator_value` returns `'apple'`, the input should be validated as an `ApplePie`, and if it + returns `'pumpkin'`, the input should be validated as a `PumpkinPie`. + + The primary role of the `Tag` here is to map the return value from the callable `Discriminator` function to + the appropriate member of the `Union` in question. + + ```python + from typing import Annotated, Any, Literal, Union + + from pydantic import BaseModel, Discriminator, Tag + + class Pie(BaseModel): + time_to_cook: int + num_ingredients: int + + class ApplePie(Pie): + fruit: Literal['apple'] = 'apple' + + class PumpkinPie(Pie): + filling: Literal['pumpkin'] = 'pumpkin' + + def get_discriminator_value(v: Any) -> str: + if isinstance(v, dict): + return v.get('fruit', v.get('filling')) + return getattr(v, 'fruit', getattr(v, 'filling', None)) + + class ThanksgivingDinner(BaseModel): + dessert: Annotated[ + Union[ + Annotated[ApplePie, Tag('apple')], + Annotated[PumpkinPie, Tag('pumpkin')], + ], + Discriminator(get_discriminator_value), + ] + + apple_variation = ThanksgivingDinner.model_validate( + {'dessert': {'fruit': 'apple', 'time_to_cook': 60, 'num_ingredients': 8}} + ) + print(repr(apple_variation)) + ''' + ThanksgivingDinner(dessert=ApplePie(time_to_cook=60, num_ingredients=8, fruit='apple')) + ''' + + pumpkin_variation = ThanksgivingDinner.model_validate( + { + 'dessert': { + 'filling': 'pumpkin', + 'time_to_cook': 40, + 'num_ingredients': 6, + } + } + ) + print(repr(pumpkin_variation)) + ''' + ThanksgivingDinner(dessert=PumpkinPie(time_to_cook=40, num_ingredients=6, filling='pumpkin')) + ''' + ``` + + !!! note + You must specify a `Tag` for every case in a `Tag` that is associated with a + callable `Discriminator`. Failing to do so will result in a `PydanticUserError` with code + [`callable-discriminator-no-tag`](../errors/usage_errors.md#callable-discriminator-no-tag). + + See the [Discriminated Unions] concepts docs for more details on how to use `Tag`s. + + [Discriminated Unions]: ../concepts/unions.md#discriminated-unions + """ + + tag: str + + def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> CoreSchema: + schema = handler(source_type) + metadata = cast('CoreMetadata', schema.setdefault('metadata', {})) + metadata['pydantic_internal_union_tag_key'] = self.tag + return schema + + +@_dataclasses.dataclass(**_internal_dataclass.slots_true, frozen=True) +class Discriminator: + """!!! abstract "Usage Documentation" + [Discriminated Unions with `Callable` `Discriminator`](../concepts/unions.md#discriminated-unions-with-callable-discriminator) + + Provides a way to use a custom callable as the way to extract the value of a union discriminator. + + This allows you to get validation behavior like you'd get from `Field(discriminator=)`, + but without needing to have a single shared field across all the union choices. This also makes it + possible to handle unions of models and primitive types with discriminated-union-style validation errors. + Finally, this allows you to use a custom callable as the way to identify which member of a union a value + belongs to, while still seeing all the performance benefits of a discriminated union. + + Consider this example, which is much more performant with the use of `Discriminator` and thus a `TaggedUnion` + than it would be as a normal `Union`. + + ```python + from typing import Annotated, Any, Literal, Union + + from pydantic import BaseModel, Discriminator, Tag + + class Pie(BaseModel): + time_to_cook: int + num_ingredients: int + + class ApplePie(Pie): + fruit: Literal['apple'] = 'apple' + + class PumpkinPie(Pie): + filling: Literal['pumpkin'] = 'pumpkin' + + def get_discriminator_value(v: Any) -> str: + if isinstance(v, dict): + return v.get('fruit', v.get('filling')) + return getattr(v, 'fruit', getattr(v, 'filling', None)) + + class ThanksgivingDinner(BaseModel): + dessert: Annotated[ + Union[ + Annotated[ApplePie, Tag('apple')], + Annotated[PumpkinPie, Tag('pumpkin')], + ], + Discriminator(get_discriminator_value), + ] + + apple_variation = ThanksgivingDinner.model_validate( + {'dessert': {'fruit': 'apple', 'time_to_cook': 60, 'num_ingredients': 8}} + ) + print(repr(apple_variation)) + ''' + ThanksgivingDinner(dessert=ApplePie(time_to_cook=60, num_ingredients=8, fruit='apple')) + ''' + + pumpkin_variation = ThanksgivingDinner.model_validate( + { + 'dessert': { + 'filling': 'pumpkin', + 'time_to_cook': 40, + 'num_ingredients': 6, + } + } + ) + print(repr(pumpkin_variation)) + ''' + ThanksgivingDinner(dessert=PumpkinPie(time_to_cook=40, num_ingredients=6, filling='pumpkin')) + ''' + ``` + + See the [Discriminated Unions] concepts docs for more details on how to use `Discriminator`s. + + [Discriminated Unions]: ../concepts/unions.md#discriminated-unions + """ + + discriminator: str | Callable[[Any], Hashable] + """The callable or field name for discriminating the type in a tagged union. + + A `Callable` discriminator must extract the value of the discriminator from the input. + A `str` discriminator must be the name of a field to discriminate against. + """ + custom_error_type: str | None = None + """Type to use in [custom errors](../errors/errors.md) replacing the standard discriminated union + validation errors. + """ + custom_error_message: str | None = None + """Message to use in custom errors.""" + custom_error_context: dict[str, int | str | float] | None = None + """Context to use in custom errors.""" + + def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> CoreSchema: + if isinstance(self.discriminator, str): + from pydantic import Field + + return handler(Annotated[source_type, Field(discriminator=self.discriminator)]) + else: + original_schema = handler(source_type) + return self._convert_schema(original_schema, handler) + + def _convert_schema( + self, original_schema: core_schema.CoreSchema, handler: GetCoreSchemaHandler | None = None + ) -> core_schema.TaggedUnionSchema: + if handler is not None and original_schema['type'] == 'definition-ref': + # Same logic as `_ApplyInferredDiscriminator._apply_to_root()` + try: + def_schema = handler.resolve_ref_schema(original_schema) + except LookupError: # pragma: no cover + from pydantic._internal._discriminated_union import MissingDefinitionForUnionRef + + raise MissingDefinitionForUnionRef(original_schema['schema_ref']) + + # If using a referenceable union as discriminated (e.g. `type Pet = Cat | Dog; field: Pet = Field(discriminator=...)`): + if def_schema['type'] == 'union': + original_schema = def_schema.copy() + original_schema.pop('ref') + + if original_schema['type'] != 'union': + # This likely indicates that the schema was a single-item union that was simplified. + # In this case, we do the same thing we do in + # `pydantic._internal._discriminated_union._ApplyInferredDiscriminator._apply_to_root`, namely, + # package the generated schema back into a single-item union. + original_schema = core_schema.union_schema([original_schema]) + + tagged_union_choices = {} + for choice in original_schema['choices']: + tag = None + if isinstance(choice, tuple): + choice, tag = choice + metadata = cast('CoreMetadata | None', choice.get('metadata')) + if metadata is not None: + tag = metadata.get('pydantic_internal_union_tag_key') or tag + if tag is None: + # `handler` is None when this method is called from `apply_discriminator()` (deferred discriminators) + if handler is not None and choice['type'] == 'definition-ref': + # If choice was built from a PEP 695 type alias, try to resolve the def: + try: + choice = handler.resolve_ref_schema(choice) + except LookupError: + pass + else: + metadata = cast('CoreMetadata | None', choice.get('metadata')) + if metadata is not None: + tag = metadata.get('pydantic_internal_union_tag_key') + + if tag is None: + raise PydanticUserError( + f'`Tag` not provided for choice {choice} used with `Discriminator`', + code='callable-discriminator-no-tag', + ) + tagged_union_choices[tag] = choice + + # Have to do these verbose checks to ensure falsy values ('' and {}) don't get ignored + custom_error_type = self.custom_error_type + if custom_error_type is None: + custom_error_type = original_schema.get('custom_error_type') + + custom_error_message = self.custom_error_message + if custom_error_message is None: + custom_error_message = original_schema.get('custom_error_message') + + custom_error_context = self.custom_error_context + if custom_error_context is None: + custom_error_context = original_schema.get('custom_error_context') + + custom_error_type = original_schema.get('custom_error_type') if custom_error_type is None else custom_error_type + return core_schema.tagged_union_schema( + tagged_union_choices, + self.discriminator, + custom_error_type=custom_error_type, + custom_error_message=custom_error_message, + custom_error_context=custom_error_context, + strict=original_schema.get('strict'), + ref=original_schema.get('ref'), + metadata=original_schema.get('metadata'), + serialization=original_schema.get('serialization'), + ) + + +_JSON_TYPES = {int, float, str, bool, list, dict, type(None)} + + +def _get_type_name(x: Any) -> str: + type_ = type(x) + if type_ in _JSON_TYPES: + return type_.__name__ + + # Handle proper subclasses; note we don't need to handle None or bool here + if isinstance(x, int): + return 'int' + if isinstance(x, float): + return 'float' + if isinstance(x, str): + return 'str' + if isinstance(x, list): + return 'list' + if isinstance(x, dict): + return 'dict' + + # Fail by returning the type's actual name + return getattr(type_, '__name__', '') + + +class _AllowAnyJson: + @classmethod + def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> CoreSchema: + python_schema = handler(source_type) + return core_schema.json_or_python_schema(json_schema=core_schema.any_schema(), python_schema=python_schema) + + +if TYPE_CHECKING: + # This seems to only be necessary for mypy + JsonValue: TypeAlias = Union[ + list['JsonValue'], + dict[str, 'JsonValue'], + str, + bool, + int, + float, + None, + ] + """A `JsonValue` is used to represent a value that can be serialized to JSON. + + It may be one of: + + * `list['JsonValue']` + * `dict[str, 'JsonValue']` + * `str` + * `bool` + * `int` + * `float` + * `None` + + The following example demonstrates how to use `JsonValue` to validate JSON data, + and what kind of errors to expect when input data is not json serializable. + + ```python + import json + + from pydantic import BaseModel, JsonValue, ValidationError + + class Model(BaseModel): + j: JsonValue + + valid_json_data = {'j': {'a': {'b': {'c': 1, 'd': [2, None]}}}} + invalid_json_data = {'j': {'a': {'b': ...}}} + + print(repr(Model.model_validate(valid_json_data))) + #> Model(j={'a': {'b': {'c': 1, 'd': [2, None]}}}) + print(repr(Model.model_validate_json(json.dumps(valid_json_data)))) + #> Model(j={'a': {'b': {'c': 1, 'd': [2, None]}}}) + + try: + Model.model_validate(invalid_json_data) + except ValidationError as e: + print(e) + ''' + 1 validation error for Model + j.dict.a.dict.b + input was not a valid JSON value [type=invalid-json-value, input_value=Ellipsis, input_type=ellipsis] + ''' + ``` + """ + +else: + JsonValue = TypeAliasType( + 'JsonValue', + Annotated[ + Union[ + Annotated[list['JsonValue'], Tag('list')], + Annotated[dict[str, 'JsonValue'], Tag('dict')], + Annotated[str, Tag('str')], + Annotated[bool, Tag('bool')], + Annotated[int, Tag('int')], + Annotated[float, Tag('float')], + Annotated[None, Tag('NoneType')], + ], + Discriminator( + _get_type_name, + custom_error_type='invalid-json-value', + custom_error_message='input was not a valid JSON value', + ), + _AllowAnyJson, + ], + ) + + +class _OnErrorOmit: + @classmethod + def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> CoreSchema: + # there is no actual default value here but we use with_default_schema since it already has the on_error + # behavior implemented and it would be no more efficient to implement it on every other validator + # or as a standalone validator + return core_schema.with_default_schema(schema=handler(source_type), on_error='omit') + + +OnErrorOmit = Annotated[T, _OnErrorOmit] +""" +When used as an item in a list, the key type in a dict, optional values of a TypedDict, etc. +this annotation omits the item from the iteration if there is any error validating it. +That is, instead of a [`ValidationError`][pydantic_core.ValidationError] being propagated up and the entire iterable being discarded +any invalid items are discarded and the valid ones are returned. +""" + + +@_dataclasses.dataclass +class FailFast(_fields.PydanticMetadata, BaseMetadata): + """A `FailFast` annotation can be used to specify that validation should stop at the first error. + + This can be useful when you want to validate a large amount of data and you only need to know if it's valid or not. + + You might want to enable this setting if you want to validate your data faster (basically, if you use this, + validation will be more performant with the caveat that you get less information). + + ```python + from typing import Annotated + + from pydantic import BaseModel, FailFast, ValidationError + + class Model(BaseModel): + x: Annotated[list[int], FailFast()] + + # This will raise a single error for the first invalid value and stop validation + try: + obj = Model(x=[1, 2, 'a', 4, 5, 'b', 7, 8, 9, 'c']) + except ValidationError as e: + print(e) + ''' + 1 validation error for Model + x.2 + Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='a', input_type=str] + ''' + ``` + """ + + fail_fast: bool = True