Загрузить файлы в «venv/Lib/site-packages/pydantic/_internal»

This commit is contained in:
2026-07-02 20:01:20 +00:00
parent c487f1f189
commit 73c35292e3
4 changed files with 1906 additions and 0 deletions

View File

@@ -0,0 +1,785 @@
"""Logic for interacting with type annotations, mostly extensions, shims and hacks to wrap Python's typing module."""
from __future__ import annotations
import collections.abc
import re
import sys
import types
import typing
from functools import partial
from inspect import Signature, signature
from typing import TYPE_CHECKING, Any, Callable, cast
import typing_extensions
from typing_extensions import deprecated, get_args, get_origin
from typing_inspection import typing_objects
from typing_inspection.introspection import is_union_origin
from pydantic.version import version_short
from ._namespace_utils import GlobalsNamespace, MappingNamespace, NsResolver, get_module_ns_of
if sys.version_info < (3, 10):
NoneType = type(None)
EllipsisType = type(Ellipsis)
else:
from types import EllipsisType as EllipsisType
from types import NoneType as NoneType
if sys.version_info >= (3, 14):
import annotationlib
if TYPE_CHECKING:
from pydantic import BaseModel
from pydantic.fields import FieldInfo
# As per https://typing-extensions.readthedocs.io/en/latest/#runtime-use-of-types,
# always check for both `typing` and `typing_extensions` variants of a typing construct.
# (this is implemented differently than the suggested approach in the `typing_extensions`
# docs for performance).
_t_annotated = typing.Annotated
_te_annotated = typing_extensions.Annotated
def is_annotated(tp: Any, /) -> bool:
"""Return whether the provided argument is a `Annotated` special form.
```python {test="skip" lint="skip"}
is_annotated(Annotated[int, ...])
#> True
```
"""
origin = get_origin(tp)
return origin is _t_annotated or origin is _te_annotated
def annotated_type(tp: Any, /) -> Any | None:
"""Return the type of the `Annotated` special form, or `None`."""
return tp.__origin__ if typing_objects.is_annotated(get_origin(tp)) else None
def unpack_type(tp: Any, /) -> Any | None:
"""Return the type wrapped by the `Unpack` special form, or `None`."""
return get_args(tp)[0] if typing_objects.is_unpack(get_origin(tp)) else None
def is_hashable(tp: Any, /) -> bool:
"""Return whether the provided argument is the `Hashable` class.
```python {test="skip" lint="skip"}
is_hashable(Hashable)
#> True
```
"""
# `get_origin` is documented as normalizing any typing-module aliases to `collections` classes,
# hence the second check:
return tp is collections.abc.Hashable or get_origin(tp) is collections.abc.Hashable
def is_callable(tp: Any, /) -> bool:
"""Return whether the provided argument is a `Callable`, parametrized or not.
```python {test="skip" lint="skip"}
is_callable(Callable[[int], str])
#> True
is_callable(typing.Callable)
#> True
is_callable(collections.abc.Callable)
#> True
```
"""
# `get_origin` is documented as normalizing any typing-module aliases to `collections` classes,
# hence the second check:
return tp is collections.abc.Callable or get_origin(tp) is collections.abc.Callable
_classvar_re = re.compile(r'((\w+\.)?Annotated\[)?(\w+\.)?ClassVar\[')
def is_classvar_annotation(tp: Any, /) -> bool:
"""Return whether the provided argument represents a class variable annotation.
Although not explicitly stated by the typing specification, `ClassVar` can be used
inside `Annotated` and as such, this function checks for this specific scenario.
Because this function is used to detect class variables before evaluating forward references
(or because evaluation failed), we also implement a naive regex match implementation. This is
required because class variables are inspected before fields are collected, so we try to be
as accurate as possible.
"""
if typing_objects.is_classvar(tp):
return True
origin = get_origin(tp)
if typing_objects.is_classvar(origin):
return True
if typing_objects.is_annotated(origin):
annotated_type = tp.__origin__
if typing_objects.is_classvar(annotated_type) or typing_objects.is_classvar(get_origin(annotated_type)):
return True
str_ann: str | None = None
if isinstance(tp, typing.ForwardRef):
str_ann = tp.__forward_arg__
if isinstance(tp, str):
str_ann = tp
if str_ann is not None and _classvar_re.match(str_ann):
# stdlib dataclasses do something similar, although a bit more advanced
# (see `dataclass._is_type`).
return True
return False
_t_final = typing.Final
_te_final = typing_extensions.Final
# TODO implement `is_finalvar_annotation` as Final can be wrapped with other special forms:
def is_finalvar(tp: Any, /) -> bool:
"""Return whether the provided argument is a `Final` special form, parametrized or not.
```python {test="skip" lint="skip"}
is_finalvar(Final[int])
#> True
is_finalvar(Final)
#> True
"""
# Final is not necessarily parametrized:
if tp is _t_final or tp is _te_final:
return True
origin = get_origin(tp)
return origin is _t_final or origin is _te_final
_NONE_TYPES: tuple[Any, ...] = (None, NoneType, typing.Literal[None], typing_extensions.Literal[None])
def is_none_type(tp: Any, /) -> bool:
"""Return whether the argument represents the `None` type as part of an annotation.
```python {test="skip" lint="skip"}
is_none_type(None)
#> True
is_none_type(NoneType)
#> True
is_none_type(Literal[None])
#> True
is_none_type(type[None])
#> False
"""
return tp in _NONE_TYPES
def is_namedtuple(tp: Any, /) -> bool:
"""Return whether the provided argument is a named tuple class.
The class can be created using `typing.NamedTuple` or `collections.namedtuple`.
Parametrized generic classes are *not* assumed to be named tuples.
"""
from ._utils import lenient_issubclass # circ. import
return lenient_issubclass(tp, tuple) and hasattr(tp, '_fields')
# TODO In 2.12, delete this export. It is currently defined only to not break
# pydantic-settings which relies on it:
origin_is_union = is_union_origin
def is_generic_alias(tp: Any, /) -> bool:
return isinstance(tp, (types.GenericAlias, typing._GenericAlias)) # pyright: ignore[reportAttributeAccessIssue]
# TODO: Ideally, we should avoid relying on the private `typing` constructs:
if sys.version_info < (3, 10):
WithArgsTypes: tuple[Any, ...] = (typing._GenericAlias, types.GenericAlias) # pyright: ignore[reportAttributeAccessIssue]
else:
WithArgsTypes: tuple[Any, ...] = (typing._GenericAlias, types.GenericAlias, types.UnionType) # pyright: ignore[reportAttributeAccessIssue]
# Similarly, we shouldn't rely on this `_Final` class, which is even more private than `_GenericAlias`:
typing_base: Any = typing._Final # pyright: ignore[reportAttributeAccessIssue]
### Annotation evaluations functions:
def parent_frame_namespace(*, parent_depth: int = 2, force: bool = False) -> dict[str, Any] | None:
"""Fetch the local namespace of the parent frame where this function is called.
Using this function is mostly useful to resolve forward annotations pointing to members defined in a local namespace,
such as assignments inside a function. Using the standard library tools, it is currently not possible to resolve
such annotations:
```python {lint="skip" test="skip"}
from typing import get_type_hints
def func() -> None:
Alias = int
class C:
a: 'Alias'
# Raises a `NameError: 'Alias' is not defined`
get_type_hints(C)
```
Pydantic uses this function when a Pydantic model is being defined to fetch the parent frame locals. However,
this only allows us to fetch the parent frame namespace and not other parents (e.g. a model defined in a function,
itself defined in another function). Inspecting the next outer frames (using `f_back`) is not reliable enough
(see https://discuss.python.org/t/20659).
Because this function is mostly used to better resolve forward annotations, nothing is returned if the parent frame's
code object is defined at the module level. In this case, the locals of the frame will be the same as the module
globals where the class is defined (see `_namespace_utils.get_module_ns_of`). However, if you still want to fetch
the module globals (e.g. when rebuilding a model, where the frame where the rebuild call is performed might contain
members that you want to use for forward annotations evaluation), you can use the `force` parameter.
Args:
parent_depth: The depth at which to get the frame. Defaults to 2, meaning the parent frame where this function
is called will be used.
force: Whether to always return the frame locals, even if the frame's code object is defined at the module level.
Returns:
The locals of the namespace, or `None` if it was skipped as per the described logic.
"""
frame = sys._getframe(parent_depth)
if frame.f_code.co_name.startswith('<generic parameters of'):
# As `parent_frame_namespace` is mostly called in `ModelMetaclass.__new__`,
# the parent frame can be the annotation scope if the PEP 695 generic syntax is used.
# (see https://docs.python.org/3/reference/executionmodel.html#annotation-scopes,
# https://docs.python.org/3/reference/compound_stmts.html#generic-classes).
# In this case, the code name is set to `<generic parameters of MyClass>`,
# and we need to skip this frame as it is irrelevant.
frame = cast(types.FrameType, frame.f_back) # guaranteed to not be `None`
# note, we don't copy frame.f_locals here (or during the last return call), because we don't expect the namespace to be
# modified down the line if this becomes a problem, we could implement some sort of frozen mapping structure to enforce this.
if force:
return frame.f_locals
# If either of the following conditions are true, the class is defined at the top module level.
# To better understand why we need both of these checks, see
# https://github.com/pydantic/pydantic/pull/10113#discussion_r1714981531.
if frame.f_back is None or frame.f_code.co_name == '<module>':
return None
return frame.f_locals
def _type_convert(arg: Any) -> Any:
"""Convert `None` to `NoneType` and strings to `ForwardRef` instances.
This is a backport of the private `typing._type_convert` function. When
evaluating a type, `ForwardRef._evaluate` ends up being called, and is
responsible for making this conversion. However, we still have to apply
it for the first argument passed to our type evaluation functions, similarly
to the `typing.get_type_hints` function.
"""
if arg is None:
return NoneType
if isinstance(arg, str):
# Like `typing.get_type_hints`, assume the arg can be in any context,
# hence the proper `is_argument` and `is_class` args:
return _make_forward_ref(arg, is_argument=False, is_class=True)
return arg
def safe_get_annotations(obj: Any) -> dict[str, Any]:
"""Get the annotations for the provided object, accounting for potential deferred forward references.
Starting with Python 3.14, accessing the `__annotations__` attribute might raise a `NameError` if
a referenced symbol isn't defined yet. In this case, we return the annotation in the *forward ref*
format.
"""
if sys.version_info >= (3, 14):
return annotationlib.get_annotations(obj, format=annotationlib.Format.FORWARDREF)
else:
# TODO just do getattr(obj, '__annotations__', {}) when dropping support for Python 3.9:
if isinstance(obj, type):
return obj.__dict__.get('__annotations__', {})
else:
return getattr(obj, '__annotations__', {})
def get_model_type_hints(
model_class: type[BaseModel],
*,
ns_resolver: NsResolver | None = None,
) -> dict[str, tuple[Any, bool]]:
"""Collect annotations from a Pydantic model class, including those from parent classes.
Args:
model_class: The Pydantic model class to inspect.
ns_resolver: A namespace resolver instance to use. Defaults to an empty instance.
Returns:
A dictionary mapping annotation names to a two-tuple: the first element is the evaluated
type or the original annotation if a `NameError` occurred, the second element is a boolean
indicating if whether the evaluation succeeded.
"""
hints: dict[str, Any] | dict[str, tuple[Any, bool]] = {}
ns_resolver = ns_resolver or NsResolver()
for base in reversed(model_class.__mro__[:-1]):
# For Python 3.14, we could also use `Format.VALUE` and pass the globals/locals
# from the ns_resolver, but we want to be able to know which specific field failed
# to evaluate:
ann = safe_get_annotations(base)
if not ann:
continue
with ns_resolver.push(base):
base_model_fields: dict[str, FieldInfo] | None = base.__dict__.get('__pydantic_fields__')
for name, value in ann.items():
if name.startswith('_'):
globalns, localns = ns_resolver.types_namespace
# For private attributes, we only need the annotation to detect the `ClassVar` special form.
# For this reason, we still try to evaluate it, but we also catch any possible exception (on
# top of the `NameError`s caught in `try_eval_type`) that could happen so that users are free
# to use any kind of forward annotation for private fields (e.g. circular imports, new typing
# syntax, etc).
try:
hints[name] = try_eval_type(value, globalns, localns)
except Exception:
hints[name] = (value, False)
else:
if base_model_fields is not None and name in base_model_fields:
# Avoid unnecessarily evaluating annotations from parent models, as we'll end up
# copying the `FieldInfo` instance from it anyway if we need to.
# We use the `annotation` attribute here, but in reality could put anything here,
# As we are guaranteed to not make use of it:
hints[name] = (base_model_fields[name].annotation, True)
else:
globalns, localns = ns_resolver.types_namespace
hints[name] = try_eval_type(value, globalns, localns)
return hints
def get_cls_type_hints(
obj: type[Any],
*,
ns_resolver: NsResolver | None = None,
) -> dict[str, Any]:
"""Collect annotations from a class, including those from parent classes.
Args:
obj: The class to inspect.
ns_resolver: A namespace resolver instance to use. Defaults to an empty instance.
"""
hints: dict[str, Any] = {}
ns_resolver = ns_resolver or NsResolver()
for base in reversed(obj.__mro__):
# For Python 3.14, we could also use `Format.VALUE` and pass the globals/locals
# from the ns_resolver, but we want to be able to know which specific field failed
# to evaluate:
ann = safe_get_annotations(base)
if not ann:
continue
with ns_resolver.push(base):
globalns, localns = ns_resolver.types_namespace
for name, value in ann.items():
hints[name] = eval_type(value, globalns, localns)
return hints
def try_eval_type(
value: Any,
globalns: GlobalsNamespace | None = None,
localns: MappingNamespace | None = None,
) -> tuple[Any, bool]:
"""Try evaluating the annotation using the provided namespaces.
Args:
value: The value to evaluate. If `None`, it will be replaced by `type[None]`. If an instance
of `str`, it will be converted to a `ForwardRef`.
localns: The global namespace to use during annotation evaluation.
globalns: The local namespace to use during annotation evaluation.
Returns:
A two-tuple containing the possibly evaluated type and a boolean indicating
whether the evaluation succeeded or not.
"""
value = _type_convert(value)
try:
return eval_type_backport(value, globalns, localns), True
except NameError:
return value, False
def eval_type(
value: Any,
globalns: GlobalsNamespace | None = None,
localns: MappingNamespace | None = None,
) -> Any:
"""Evaluate the annotation using the provided namespaces.
Args:
value: The value to evaluate. If `None`, it will be replaced by `type[None]`. If an instance
of `str`, it will be converted to a `ForwardRef`.
localns: The global namespace to use during annotation evaluation.
globalns: The local namespace to use during annotation evaluation.
"""
value = _type_convert(value)
return eval_type_backport(value, globalns, localns)
@deprecated(
'`eval_type_lenient` is deprecated, use `try_eval_type` instead.',
category=None,
)
def eval_type_lenient(
value: Any,
globalns: GlobalsNamespace | None = None,
localns: MappingNamespace | None = None,
) -> Any:
ev, _ = try_eval_type(value, globalns, localns)
return ev
def eval_type_backport(
value: Any,
globalns: GlobalsNamespace | None = None,
localns: MappingNamespace | None = None,
type_params: tuple[Any, ...] | None = None,
) -> Any:
"""An enhanced version of `typing._eval_type` which will fall back to using the `eval_type_backport`
package if it's installed to let older Python versions use newer typing constructs.
Specifically, this transforms `X | Y` into `typing.Union[X, Y]` and `list[X]` into `typing.List[X]`
(as well as all the types made generic in PEP 585) if the original syntax is not supported in the
current Python version.
This function will also display a helpful error if the value passed fails to evaluate.
"""
try:
return _eval_type_backport(value, globalns, localns, type_params)
except TypeError as e:
if 'Unable to evaluate type annotation' in str(e):
raise
# If it is a `TypeError` and value isn't a `ForwardRef`, it would have failed during annotation definition.
# Thus we assert here for type checking purposes:
assert isinstance(value, typing.ForwardRef)
message = f'Unable to evaluate type annotation {value.__forward_arg__!r}.'
if sys.version_info >= (3, 11):
e.add_note(message)
raise
else:
raise TypeError(message) from e
except RecursionError as e:
# TODO ideally recursion errors should be checked in `eval_type` above, but `eval_type_backport`
# is used directly in some places.
message = (
"If you made use of an implicit recursive type alias (e.g. `MyType = list['MyType']), "
'consider using PEP 695 type aliases instead. For more details, refer to the documentation: '
f'https://docs.pydantic.dev/{version_short()}/concepts/types/#named-recursive-types'
)
if sys.version_info >= (3, 11):
e.add_note(message)
raise
else:
raise RecursionError(f'{e.args[0]}\n{message}')
def _eval_type_backport(
value: Any,
globalns: GlobalsNamespace | None = None,
localns: MappingNamespace | None = None,
type_params: tuple[Any, ...] | None = None,
) -> Any:
try:
return _eval_type(value, globalns, localns, type_params)
except TypeError as e:
if not (isinstance(value, typing.ForwardRef) and is_backport_fixable_error(e)):
raise
try:
from eval_type_backport import eval_type_backport
except ImportError:
raise TypeError(
f'Unable to evaluate type annotation {value.__forward_arg__!r}. If you are making use '
'of the new typing syntax (unions using `|` since Python 3.10 or builtins subscripting '
'since Python 3.9), you should either replace the use of new syntax with the existing '
'`typing` constructs or install the `eval_type_backport` package.'
) from e
return eval_type_backport(
value,
globalns,
localns, # pyright: ignore[reportArgumentType], waiting on a new `eval_type_backport` release.
try_default=False,
)
def _eval_type(
value: Any,
globalns: GlobalsNamespace | None = None,
localns: MappingNamespace | None = None,
type_params: tuple[Any, ...] | None = None,
) -> Any:
if sys.version_info >= (3, 14):
# Starting in 3.14, `_eval_type()` does *not* apply `_type_convert()`
# anymore. This means the `None` -> `type(None)` conversion does not apply:
evaluated = typing._eval_type( # type: ignore
value,
globalns,
localns,
type_params=type_params,
# This is relevant when evaluating types from `TypedDict` classes, where string annotations
# are automatically converted to `ForwardRef` instances with a module set. In this case,
# Our `globalns` is irrelevant and we need to indicate `typing._eval_type()` that it should
# infer it from the `ForwardRef.__forward_module__` attribute instead (`typing.get_type_hints()`
# does the same). Note that this would probably be unnecessary if we properly iterated over the
# `__orig_bases__` for TypedDicts in `get_cls_type_hints()`:
prefer_fwd_module=True,
)
if evaluated is None:
evaluated = type(None)
return evaluated
elif sys.version_info >= (3, 13):
return typing._eval_type( # type: ignore
value, globalns, localns, type_params=type_params
)
else:
return typing._eval_type( # type: ignore
value, globalns, localns
)
def is_backport_fixable_error(e: TypeError) -> bool:
msg = str(e)
return sys.version_info < (3, 10) and msg.startswith('unsupported operand type(s) for |: ')
def signature_no_eval(f: Callable[..., Any]) -> Signature:
"""Get the signature of a callable without evaluating any annotations."""
if sys.version_info >= (3, 14):
from annotationlib import Format
return signature(f, annotation_format=Format.FORWARDREF)
else:
return signature(f)
def get_function_type_hints(
function: Callable[..., Any],
*,
include_keys: set[str] | None = None,
globalns: GlobalsNamespace | None = None,
localns: MappingNamespace | None = None,
) -> dict[str, Any]:
"""Return type hints for a function.
This is similar to the `typing.get_type_hints` function, with a few differences:
- Support `functools.partial` by using the underlying `func` attribute.
- Do not wrap type annotation of a parameter with `Optional` if it has a default value of `None`
(related bug: https://github.com/python/cpython/issues/90353, only fixed in 3.11+).
"""
if isinstance(function, partial):
annotations = safe_get_annotations(function.func)
else:
annotations = safe_get_annotations(function)
if globalns is None:
globalns = get_module_ns_of(function)
type_params: tuple[Any, ...] | None = None
if localns is None:
# If localns was specified, it is assumed to already contain type params. This is because
# Pydantic has more advanced logic to do so (see `_namespace_utils.ns_for_function`).
type_params = getattr(function, '__type_params__', ())
type_hints = {}
for name, value in annotations.items():
if include_keys is not None and name not in include_keys:
continue
if value is None:
value = NoneType
elif isinstance(value, str):
value = _make_forward_ref(value)
type_hints[name] = eval_type_backport(value, globalns, localns, type_params)
return type_hints
# TODO use typing.ForwardRef directly when we stop supporting 3.9:
if sys.version_info < (3, 9, 8) or (3, 10) <= sys.version_info < (3, 10, 1):
def _make_forward_ref(
arg: Any,
is_argument: bool = True,
*,
is_class: bool = False,
) -> typing.ForwardRef:
"""Wrapper for ForwardRef that accounts for the `is_class` argument missing in older versions.
The `module` argument is omitted as it breaks <3.9.8, =3.10.0 and isn't used in the calls below.
See https://github.com/python/cpython/pull/28560 for some background.
The backport happened on 3.9.8, see:
https://github.com/pydantic/pydantic/discussions/6244#discussioncomment-6275458,
and on 3.10.1 for the 3.10 branch, see:
https://github.com/pydantic/pydantic/issues/6912
Implemented as EAFP with memory.
"""
return typing.ForwardRef(arg, is_argument) # pyright: ignore[reportCallIssue]
else:
_make_forward_ref = typing.ForwardRef # pyright: ignore[reportAssignmentType]
if sys.version_info >= (3, 10):
get_type_hints = typing.get_type_hints
else:
"""
For older versions of python, we have a custom implementation of `get_type_hints` which is a close as possible to
the implementation in CPython 3.10.8.
"""
@typing.no_type_check
def get_type_hints( # noqa: C901
obj: Any,
globalns: dict[str, Any] | None = None,
localns: dict[str, Any] | None = None,
include_extras: bool = False,
) -> dict[str, Any]: # pragma: no cover
"""Taken verbatim from python 3.10.8 unchanged, except:
* type annotations of the function definition above.
* prefixing `typing.` where appropriate
* Use `_make_forward_ref` instead of `typing.ForwardRef` to handle the `is_class` argument.
https://github.com/python/cpython/blob/aaaf5174241496afca7ce4d4584570190ff972fe/Lib/typing.py#L1773-L1875
DO NOT CHANGE THIS METHOD UNLESS ABSOLUTELY NECESSARY.
======================================================
Return type hints for an object.
This is often the same as obj.__annotations__, but it handles
forward references encoded as string literals, adds Optional[t] if a
default value equal to None is set and recursively replaces all
'Annotated[T, ...]' with 'T' (unless 'include_extras=True').
The argument may be a module, class, method, or function. The annotations
are returned as a dictionary. For classes, annotations include also
inherited members.
TypeError is raised if the argument is not of a type that can contain
annotations, and an empty dictionary is returned if no annotations are
present.
BEWARE -- the behavior of globalns and localns is counterintuitive
(unless you are familiar with how eval() and exec() work). The
search order is locals first, then globals.
- If no dict arguments are passed, an attempt is made to use the
globals from obj (or the respective module's globals for classes),
and these are also used as the locals. If the object does not appear
to have globals, an empty dictionary is used. For classes, the search
order is globals first then locals.
- If one dict argument is passed, it is used for both globals and
locals.
- If two dict arguments are passed, they specify globals and
locals, respectively.
"""
if getattr(obj, '__no_type_check__', None):
return {}
# Classes require a special treatment.
if isinstance(obj, type):
hints = {}
for base in reversed(obj.__mro__):
if globalns is None:
base_globals = getattr(sys.modules.get(base.__module__, None), '__dict__', {})
else:
base_globals = globalns
ann = base.__dict__.get('__annotations__', {})
if isinstance(ann, types.GetSetDescriptorType):
ann = {}
base_locals = dict(vars(base)) if localns is None else localns
if localns is None and globalns is None:
# This is surprising, but required. Before Python 3.10,
# get_type_hints only evaluated the globalns of
# a class. To maintain backwards compatibility, we reverse
# the globalns and localns order so that eval() looks into
# *base_globals* first rather than *base_locals*.
# This only affects ForwardRefs.
base_globals, base_locals = base_locals, base_globals
for name, value in ann.items():
if value is None:
value = type(None)
if isinstance(value, str):
value = _make_forward_ref(value, is_argument=False, is_class=True)
value = eval_type_backport(value, base_globals, base_locals)
hints[name] = value
if not include_extras and hasattr(typing, '_strip_annotations'):
return {
k: typing._strip_annotations(t) # type: ignore
for k, t in hints.items()
}
else:
return hints
if globalns is None:
if isinstance(obj, types.ModuleType):
globalns = obj.__dict__
else:
nsobj = obj
# Find globalns for the unwrapped object.
while hasattr(nsobj, '__wrapped__'):
nsobj = nsobj.__wrapped__
globalns = getattr(nsobj, '__globals__', {})
if localns is None:
localns = globalns
elif localns is None:
localns = globalns
hints = getattr(obj, '__annotations__', None)
if hints is None:
# Return empty annotations for something that _could_ have them.
if isinstance(obj, typing._allowed_types): # type: ignore
return {}
else:
raise TypeError(f'{obj!r} is not a module, class, method, or function.')
defaults = typing._get_defaults(obj) # type: ignore
hints = dict(hints)
for name, value in hints.items():
if value is None:
value = type(None)
if isinstance(value, str):
# class-level forward refs were handled above, this must be either
# a module-level annotation or a function argument annotation
value = _make_forward_ref(
value,
is_argument=not isinstance(obj, types.ModuleType),
is_class=False,
)
value = eval_type_backport(value, globalns, localns)
if name in defaults and defaults[name] is None:
value = typing.Optional[value]
hints[name] = value
return hints if include_extras else {k: typing._strip_annotations(t) for k, t in hints.items()} # type: ignore

View File

@@ -0,0 +1,446 @@
"""Bucket of reusable internal utilities.
This should be reduced as much as possible with functions only used in one place, moved to that place.
"""
from __future__ import annotations as _annotations
import dataclasses
import keyword
import sys
import warnings
import weakref
from collections import OrderedDict, defaultdict, deque
from collections.abc import Callable, Iterable, Mapping
from collections.abc import Set as AbstractSet
from copy import deepcopy
from functools import cached_property
from inspect import Parameter
from itertools import zip_longest
from types import BuiltinFunctionType, CodeType, FunctionType, GeneratorType, LambdaType, ModuleType
from typing import TYPE_CHECKING, Any, Generic, TypeVar, overload
from pydantic_core import MISSING, PydanticUndefined
from typing_extensions import TypeAlias, TypeGuard, deprecated
from pydantic import PydanticDeprecatedSince211
from . import _repr, _typing_extra
from ._import_utils import import_cached_base_model
if TYPE_CHECKING:
# TODO remove type error comments when we drop support for Python 3.9
MappingIntStrAny: TypeAlias = Mapping[int, Any] | Mapping[str, Any] # pyright: ignore[reportGeneralTypeIssues]
AbstractSetIntStr: TypeAlias = AbstractSet[int] | AbstractSet[str] # pyright: ignore[reportGeneralTypeIssues]
from ..main import BaseModel
# these are types that are returned unchanged by deepcopy
IMMUTABLE_NON_COLLECTIONS_TYPES: set[type[Any]] = {
int,
float,
complex,
str,
bool,
bytes,
type,
_typing_extra.NoneType,
FunctionType,
BuiltinFunctionType,
LambdaType,
weakref.ref,
CodeType,
# note: including ModuleType will differ from behaviour of deepcopy by not producing error.
# It might be not a good idea in general, but considering that this function used only internally
# against default values of fields, this will allow to actually have a field with module as default value
ModuleType,
NotImplemented.__class__,
Ellipsis.__class__,
}
# these are types that if empty, might be copied with simple copy() instead of deepcopy()
BUILTIN_COLLECTIONS: set[type[Any]] = {
list,
set,
tuple,
frozenset,
dict,
OrderedDict,
defaultdict,
deque,
}
def can_be_positional(param: Parameter) -> bool:
"""Return whether the parameter accepts a positional argument.
```python {test="skip" lint="skip"}
def func(a, /, b, *, c):
pass
params = inspect.signature(func).parameters
can_be_positional(params['a'])
#> True
can_be_positional(params['b'])
#> True
can_be_positional(params['c'])
#> False
```
"""
return param.kind in (Parameter.POSITIONAL_ONLY, Parameter.POSITIONAL_OR_KEYWORD)
def sequence_like(v: Any) -> bool:
return isinstance(v, (list, tuple, set, frozenset, GeneratorType, deque))
def lenient_isinstance(o: Any, class_or_tuple: type[Any] | tuple[type[Any], ...] | None) -> bool: # pragma: no cover
try:
return isinstance(o, class_or_tuple) # type: ignore[arg-type]
except TypeError:
return False
def lenient_issubclass(cls: Any, class_or_tuple: Any) -> bool: # pragma: no cover
try:
return isinstance(cls, type) and issubclass(cls, class_or_tuple)
except TypeError:
if isinstance(cls, _typing_extra.WithArgsTypes):
return False
raise # pragma: no cover
def is_model_class(cls: Any) -> TypeGuard[type[BaseModel]]:
"""Returns true if cls is a _proper_ subclass of BaseModel, and provides proper type-checking,
unlike raw calls to lenient_issubclass.
"""
BaseModel = import_cached_base_model()
return lenient_issubclass(cls, BaseModel) and cls is not BaseModel
def is_valid_identifier(identifier: str) -> bool:
"""Checks that a string is a valid identifier and not a Python keyword.
:param identifier: The identifier to test.
:return: True if the identifier is valid.
"""
return identifier.isidentifier() and not keyword.iskeyword(identifier)
KeyType = TypeVar('KeyType')
def deep_update(mapping: dict[KeyType, Any], *updating_mappings: dict[KeyType, Any]) -> dict[KeyType, Any]:
updated_mapping = mapping.copy()
for updating_mapping in updating_mappings:
for k, v in updating_mapping.items():
if k in updated_mapping and isinstance(updated_mapping[k], dict) and isinstance(v, dict):
updated_mapping[k] = deep_update(updated_mapping[k], v)
else:
updated_mapping[k] = v
return updated_mapping
def update_not_none(mapping: dict[Any, Any], **update: Any) -> None:
mapping.update({k: v for k, v in update.items() if v is not None})
T = TypeVar('T')
def unique_list(
input_list: list[T] | tuple[T, ...],
*,
name_factory: Callable[[T], str] = str,
) -> list[T]:
"""Make a list unique while maintaining order.
We update the list if another one with the same name is set
(e.g. model validator overridden in subclass).
"""
result: list[T] = []
result_names: list[str] = []
for v in input_list:
v_name = name_factory(v)
if v_name not in result_names:
result_names.append(v_name)
result.append(v)
else:
result[result_names.index(v_name)] = v
return result
class ValueItems(_repr.Representation):
"""Class for more convenient calculation of excluded or included fields on values."""
__slots__ = ('_items', '_type')
def __init__(self, value: Any, items: AbstractSetIntStr | MappingIntStrAny) -> None:
items = self._coerce_items(items)
if isinstance(value, (list, tuple)):
items = self._normalize_indexes(items, len(value)) # type: ignore
self._items: MappingIntStrAny = items # type: ignore
def is_excluded(self, item: Any) -> bool:
"""Check if item is fully excluded.
:param item: key or index of a value
"""
return self.is_true(self._items.get(item))
def is_included(self, item: Any) -> bool:
"""Check if value is contained in self._items.
:param item: key or index of value
"""
return item in self._items
def for_element(self, e: int | str) -> AbstractSetIntStr | MappingIntStrAny | None:
""":param e: key or index of element on value
:return: raw values for element if self._items is dict and contain needed element
"""
item = self._items.get(e) # type: ignore
return item if not self.is_true(item) else None
def _normalize_indexes(self, items: MappingIntStrAny, v_length: int) -> dict[int | str, Any]:
""":param items: dict or set of indexes which will be normalized
:param v_length: length of sequence indexes of which will be
>>> self._normalize_indexes({0: True, -2: True, -1: True}, 4)
{0: True, 2: True, 3: True}
>>> self._normalize_indexes({'__all__': True}, 4)
{0: True, 1: True, 2: True, 3: True}
"""
normalized_items: dict[int | str, Any] = {}
all_items = None
for i, v in items.items():
if not (isinstance(v, Mapping) or isinstance(v, AbstractSet) or self.is_true(v)):
raise TypeError(f'Unexpected type of exclude value for index "{i}" {v.__class__}')
if i == '__all__':
all_items = self._coerce_value(v)
continue
if not isinstance(i, int):
raise TypeError(
'Excluding fields from a sequence of sub-models or dicts must be performed index-wise: '
'expected integer keys or keyword "__all__"'
)
normalized_i = v_length + i if i < 0 else i
normalized_items[normalized_i] = self.merge(v, normalized_items.get(normalized_i))
if not all_items:
return normalized_items
if self.is_true(all_items):
for i in range(v_length):
normalized_items.setdefault(i, ...)
return normalized_items
for i in range(v_length):
normalized_item = normalized_items.setdefault(i, {})
if not self.is_true(normalized_item):
normalized_items[i] = self.merge(all_items, normalized_item)
return normalized_items
@classmethod
def merge(cls, base: Any, override: Any, intersect: bool = False) -> Any:
"""Merge a `base` item with an `override` item.
Both `base` and `override` are converted to dictionaries if possible.
Sets are converted to dictionaries with the sets entries as keys and
Ellipsis as values.
Each key-value pair existing in `base` is merged with `override`,
while the rest of the key-value pairs are updated recursively with this function.
Merging takes place based on the "union" of keys if `intersect` is
set to `False` (default) and on the intersection of keys if
`intersect` is set to `True`.
"""
override = cls._coerce_value(override)
base = cls._coerce_value(base)
if override is None:
return base
if cls.is_true(base) or base is None:
return override
if cls.is_true(override):
return base if intersect else override
# intersection or union of keys while preserving ordering:
if intersect:
merge_keys = [k for k in base if k in override] + [k for k in override if k in base]
else:
merge_keys = list(base) + [k for k in override if k not in base]
merged: dict[int | str, Any] = {}
for k in merge_keys:
merged_item = cls.merge(base.get(k), override.get(k), intersect=intersect)
if merged_item is not None:
merged[k] = merged_item
return merged
@staticmethod
def _coerce_items(items: AbstractSetIntStr | MappingIntStrAny) -> MappingIntStrAny:
if isinstance(items, Mapping):
pass
elif isinstance(items, AbstractSet):
items = dict.fromkeys(items, ...) # type: ignore
else:
class_name = getattr(items, '__class__', '???')
raise TypeError(f'Unexpected type of exclude value {class_name}')
return items # type: ignore
@classmethod
def _coerce_value(cls, value: Any) -> Any:
if value is None or cls.is_true(value):
return value
return cls._coerce_items(value)
@staticmethod
def is_true(v: Any) -> bool:
return v is True or v is ...
def __repr_args__(self) -> _repr.ReprArgs:
return [(None, self._items)]
if TYPE_CHECKING:
def LazyClassAttribute(name: str, get_value: Callable[[], T]) -> T: ...
else:
class LazyClassAttribute:
"""A descriptor exposing an attribute only accessible on a class (hidden from instances).
The attribute is lazily computed and cached during the first access.
"""
def __init__(self, name: str, get_value: Callable[[], Any]) -> None:
self.name = name
self.get_value = get_value
@cached_property
def value(self) -> Any:
return self.get_value()
def __get__(self, instance: Any, owner: type[Any]) -> None:
if instance is None:
return self.value
raise AttributeError(f'{self.name!r} attribute of {owner.__name__!r} is class-only')
Obj = TypeVar('Obj')
def smart_deepcopy(obj: Obj) -> Obj:
"""Return type as is for immutable built-in types
Use obj.copy() for built-in empty collections
Use copy.deepcopy() for non-empty collections and unknown objects.
"""
if obj is MISSING or obj is PydanticUndefined:
return obj # pyright: ignore[reportReturnType]
obj_type = obj.__class__
if obj_type in IMMUTABLE_NON_COLLECTIONS_TYPES:
return obj # fastest case: obj is immutable and not collection therefore will not be copied anyway
try:
if not obj and obj_type in BUILTIN_COLLECTIONS:
# faster way for empty collections, no need to copy its members
return obj if obj_type is tuple else obj.copy() # tuple doesn't have copy method # type: ignore
except (TypeError, ValueError, RuntimeError):
# do we really dare to catch ALL errors? Seems a bit risky
pass
return deepcopy(obj) # slowest way when we actually might need a deepcopy
_SENTINEL = object()
def all_identical(left: Iterable[Any], right: Iterable[Any]) -> bool:
"""Check that the items of `left` are the same objects as those in `right`.
>>> a, b = object(), object()
>>> all_identical([a, b, a], [a, b, a])
True
>>> all_identical([a, b, [a]], [a, b, [a]]) # new list object, while "equal" is not "identical"
False
"""
for left_item, right_item in zip_longest(left, right, fillvalue=_SENTINEL):
if left_item is not right_item:
return False
return True
def get_first_not_none(a: Any, b: Any) -> Any:
"""Return the first argument if it is not `None`, otherwise return the second argument."""
return a if a is not None else b
@dataclasses.dataclass(frozen=True)
class SafeGetItemProxy:
"""Wrapper redirecting `__getitem__` to `get` with a sentinel value as default
This makes is safe to use in `operator.itemgetter` when some keys may be missing
"""
# Define __slots__manually for performances
# @dataclasses.dataclass() only support slots=True in python>=3.10
__slots__ = ('wrapped',)
wrapped: Mapping[str, Any]
def __getitem__(self, key: str, /) -> Any:
return self.wrapped.get(key, _SENTINEL)
# required to pass the object to operator.itemgetter() instances due to a quirk of typeshed
# https://github.com/python/mypy/issues/13713
# https://github.com/python/typeshed/pull/8785
# Since this is typing-only, hide it in a typing.TYPE_CHECKING block
if TYPE_CHECKING:
def __contains__(self, key: str, /) -> bool:
return self.wrapped.__contains__(key)
_ModelT = TypeVar('_ModelT', bound='BaseModel')
_RT = TypeVar('_RT')
class deprecated_instance_property(Generic[_ModelT, _RT]):
"""A decorator exposing the decorated class method as a property, with a warning on instance access.
This decorator takes a class method defined on the `BaseModel` class and transforms it into
an attribute. The attribute can be accessed on both the class and instances of the class. If accessed
via an instance, a deprecation warning is emitted stating that instance access will be removed in V3.
"""
def __init__(self, fget: Callable[[type[_ModelT]], _RT], /) -> None:
# Note: fget should be a classmethod:
self.fget = fget
@overload
def __get__(self, instance: None, objtype: type[_ModelT]) -> _RT: ...
@overload
@deprecated(
'Accessing this attribute on the instance is deprecated, and will be removed in Pydantic V3. '
'Instead, you should access this attribute from the model class.',
category=None,
)
def __get__(self, instance: _ModelT, objtype: type[_ModelT]) -> _RT: ...
def __get__(self, instance: _ModelT | None, objtype: type[_ModelT]) -> _RT:
if instance is not None:
# fmt: off
attr_name = (
self.fget.__name__
if sys.version_info >= (3, 10)
else self.fget.__func__.__name__ # pyright: ignore[reportFunctionMemberAccess]
)
# fmt: on
warnings.warn(
f'Accessing the {attr_name!r} attribute on the instance is deprecated. '
'Instead, you should access this attribute from the model class.',
category=PydanticDeprecatedSince211,
stacklevel=2,
)
return self.fget.__get__(instance, objtype)()

View File

@@ -0,0 +1,141 @@
from __future__ import annotations as _annotations
import functools
import inspect
from collections.abc import Awaitable
from functools import partial
from typing import Any, Callable
import pydantic_core
from ..config import ConfigDict
from ..plugin._schema_validator import create_schema_validator
from ._config import ConfigWrapper
from ._generate_schema import GenerateSchema, ValidateCallSupportedTypes
from ._namespace_utils import MappingNamespace, NsResolver, ns_for_function
from ._typing_extra import signature_no_eval
def extract_function_name(func: ValidateCallSupportedTypes) -> str:
"""Extract the name of a `ValidateCallSupportedTypes` object."""
return f'partial({func.func.__name__})' if isinstance(func, functools.partial) else func.__name__
def extract_function_qualname(func: ValidateCallSupportedTypes) -> str:
"""Extract the qualname of a `ValidateCallSupportedTypes` object."""
return f'partial({func.func.__qualname__})' if isinstance(func, functools.partial) else func.__qualname__
def update_wrapper_attributes(wrapped: ValidateCallSupportedTypes, wrapper: Callable[..., Any]):
"""Update the `wrapper` function with the attributes of the `wrapped` function. Return the updated function."""
if inspect.iscoroutinefunction(wrapped):
@functools.wraps(wrapped)
async def wrapper_function(*args, **kwargs): # type: ignore
return await wrapper(*args, **kwargs)
else:
@functools.wraps(wrapped)
def wrapper_function(*args, **kwargs):
return wrapper(*args, **kwargs)
# We need to manually update this because `partial` object has no `__name__` and `__qualname__`.
wrapper_function.__name__ = extract_function_name(wrapped)
wrapper_function.__qualname__ = extract_function_qualname(wrapped)
wrapper_function.raw_function = wrapped # type: ignore
return wrapper_function
class ValidateCallWrapper:
"""This is a wrapper around a function that validates the arguments passed to it, and optionally the return value."""
__slots__ = (
'function',
'validate_return',
'schema_type',
'module',
'qualname',
'ns_resolver',
'config_wrapper',
'__pydantic_complete__',
'__pydantic_validator__',
'__return_pydantic_validator__',
)
def __init__(
self,
function: ValidateCallSupportedTypes,
config: ConfigDict | None,
validate_return: bool,
parent_namespace: MappingNamespace | None,
) -> None:
self.function = function
self.validate_return = validate_return
if isinstance(function, partial):
self.schema_type = function.func
self.module = function.func.__module__
else:
self.schema_type = function
self.module = function.__module__
self.qualname = extract_function_qualname(function)
self.ns_resolver = NsResolver(
namespaces_tuple=ns_for_function(self.schema_type, parent_namespace=parent_namespace)
)
self.config_wrapper = ConfigWrapper(config)
if not self.config_wrapper.defer_build:
self._create_validators()
else:
self.__pydantic_complete__ = False
def _create_validators(self) -> None:
gen_schema = GenerateSchema(self.config_wrapper, self.ns_resolver)
schema = gen_schema.clean_schema(gen_schema.generate_schema(self.function))
core_config = self.config_wrapper.core_config(title=self.qualname)
self.__pydantic_validator__ = create_schema_validator(
schema,
self.schema_type,
self.module,
self.qualname,
'validate_call',
core_config,
self.config_wrapper.plugin_settings,
)
if self.validate_return:
signature = signature_no_eval(self.function)
return_type = signature.return_annotation if signature.return_annotation is not signature.empty else Any
gen_schema = GenerateSchema(self.config_wrapper, self.ns_resolver)
schema = gen_schema.clean_schema(gen_schema.generate_schema(return_type))
validator = create_schema_validator(
schema,
self.schema_type,
self.module,
self.qualname,
'validate_call',
core_config,
self.config_wrapper.plugin_settings,
)
if inspect.iscoroutinefunction(self.function):
async def return_val_wrapper(aw: Awaitable[Any]) -> None:
return validator.validate_python(await aw)
self.__return_pydantic_validator__ = return_val_wrapper
else:
self.__return_pydantic_validator__ = validator.validate_python
else:
self.__return_pydantic_validator__ = None
self.__pydantic_complete__ = True
def __call__(self, *args: Any, **kwargs: Any) -> Any:
if not self.__pydantic_complete__:
self._create_validators()
res = self.__pydantic_validator__.validate_python(pydantic_core.ArgsKwargs(args, kwargs))
if self.__return_pydantic_validator__:
return self.__return_pydantic_validator__(res)
else:
return res

View File

@@ -0,0 +1,534 @@
"""Validator functions for standard library types.
Import of this module is deferred since it contains imports of many standard library modules.
"""
from __future__ import annotations as _annotations
import collections.abc
import math
import re
import typing
from collections.abc import Sequence
from decimal import Decimal
from fractions import Fraction
from ipaddress import IPv4Address, IPv4Interface, IPv4Network, IPv6Address, IPv6Interface, IPv6Network
from typing import Any, Callable, TypeVar, Union, cast
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
import typing_extensions
from pydantic_core import PydanticCustomError, PydanticKnownError, core_schema
from typing_extensions import get_args, get_origin
from typing_inspection import typing_objects
from pydantic._internal._import_utils import import_cached_field_info
from pydantic.errors import PydanticSchemaGenerationError
def sequence_validator(
input_value: Sequence[Any],
/,
validator: core_schema.ValidatorFunctionWrapHandler,
) -> Sequence[Any]:
"""Validator for `Sequence` types, isinstance(v, Sequence) has already been called."""
value_type = type(input_value)
# We don't accept any plain string as a sequence
# Relevant issue: https://github.com/pydantic/pydantic/issues/5595
if issubclass(value_type, (str, bytes)):
raise PydanticCustomError(
'sequence_str',
"'{type_name}' instances are not allowed as a Sequence value",
{'type_name': value_type.__name__},
)
# TODO: refactor sequence validation to validate with either a list or a tuple
# schema, depending on the type of the value.
# Additionally, we should be able to remove one of either this validator or the
# SequenceValidator in _std_types_schema.py (preferably this one, while porting over some logic).
# Effectively, a refactor for sequence validation is needed.
if value_type is tuple:
input_value = list(input_value)
v_list = validator(input_value)
# the rest of the logic is just re-creating the original type from `v_list`
if value_type is list:
return v_list
elif issubclass(value_type, range):
# return the list as we probably can't re-create the range
return v_list
elif value_type is tuple:
return tuple(v_list)
else:
# best guess at how to re-create the original type, more custom construction logic might be required
return value_type(v_list) # type: ignore[call-arg]
def import_string(value: Any) -> Any:
if isinstance(value, str):
try:
return _import_string_logic(value)
except ImportError as e:
raise PydanticCustomError('import_error', 'Invalid python path: {error}', {'error': str(e)}) from e
else:
# otherwise we just return the value and let the next validator do the rest of the work
return value
def _import_string_logic(dotted_path: str) -> Any:
"""Inspired by uvicorn — dotted paths should include a colon before the final item if that item is not a module.
(This is necessary to distinguish between a submodule and an attribute when there is a conflict.).
If the dotted path does not include a colon and the final item is not a valid module, importing as an attribute
rather than a submodule will be attempted automatically.
So, for example, the following values of `dotted_path` result in the following returned values:
* 'collections': <module 'collections'>
* 'collections.abc': <module 'collections.abc'>
* 'collections.abc:Mapping': <class 'collections.abc.Mapping'>
* `collections.abc.Mapping`: <class 'collections.abc.Mapping'> (though this is a bit slower than the previous line)
An error will be raised under any of the following scenarios:
* `dotted_path` contains more than one colon (e.g., 'collections:abc:Mapping')
* the substring of `dotted_path` before the colon is not a valid module in the environment (e.g., '123:Mapping')
* the substring of `dotted_path` after the colon is not an attribute of the module (e.g., 'collections:abc123')
"""
from importlib import import_module
components = dotted_path.strip().split(':')
if len(components) > 2:
raise ImportError(f"Import strings should have at most one ':'; received {dotted_path!r}")
attribute = None
if len(components) == 2:
attribute = components[1]
module_path = components[0]
if not module_path:
raise ImportError(f'Import strings should have a nonempty module name; received {dotted_path!r}')
try:
module = import_module(module_path)
except ModuleNotFoundError:
if attribute is None and '.' in module_path:
# Try interpreting the final dotted segment as an attribute, not a submodule
maybe_module_path, maybe_attribute = module_path.rsplit('.', 1)
try:
return _import_string_logic(f'{maybe_module_path}:{maybe_attribute}')
except ImportError:
pass
raise
if attribute is not None:
try:
return getattr(module, attribute)
except AttributeError as e:
raise ImportError(f'cannot import name {attribute!r} from {module_path!r}') from e
else:
return module
def pattern_either_validator(input_value: Any, /) -> re.Pattern[Any]:
if isinstance(input_value, re.Pattern):
return input_value
elif isinstance(input_value, (str, bytes)):
# todo strict mode
return compile_pattern(input_value) # type: ignore
else:
raise PydanticCustomError('pattern_type', 'Input should be a valid pattern')
def pattern_str_validator(input_value: Any, /) -> re.Pattern[str]:
if isinstance(input_value, re.Pattern):
if isinstance(input_value.pattern, str):
return input_value
else:
raise PydanticCustomError('pattern_str_type', 'Input should be a string pattern')
elif isinstance(input_value, str):
return compile_pattern(input_value)
elif isinstance(input_value, bytes):
raise PydanticCustomError('pattern_str_type', 'Input should be a string pattern')
else:
raise PydanticCustomError('pattern_type', 'Input should be a valid pattern')
def pattern_bytes_validator(input_value: Any, /) -> re.Pattern[bytes]:
if isinstance(input_value, re.Pattern):
if isinstance(input_value.pattern, bytes):
return input_value
else:
raise PydanticCustomError('pattern_bytes_type', 'Input should be a bytes pattern')
elif isinstance(input_value, bytes):
return compile_pattern(input_value)
elif isinstance(input_value, str):
raise PydanticCustomError('pattern_bytes_type', 'Input should be a bytes pattern')
else:
raise PydanticCustomError('pattern_type', 'Input should be a valid pattern')
PatternType = TypeVar('PatternType', str, bytes)
def compile_pattern(pattern: PatternType) -> re.Pattern[PatternType]:
try:
return re.compile(pattern)
except re.error:
raise PydanticCustomError('pattern_regex', 'Input should be a valid regular expression')
def ip_v4_address_validator(input_value: Any, /) -> IPv4Address:
if isinstance(input_value, IPv4Address):
return input_value
try:
return IPv4Address(input_value)
except ValueError:
raise PydanticCustomError('ip_v4_address', 'Input is not a valid IPv4 address')
def ip_v6_address_validator(input_value: Any, /) -> IPv6Address:
if isinstance(input_value, IPv6Address):
return input_value
try:
return IPv6Address(input_value)
except ValueError:
raise PydanticCustomError('ip_v6_address', 'Input is not a valid IPv6 address')
def ip_v4_network_validator(input_value: Any, /) -> IPv4Network:
"""Assume IPv4Network initialised with a default `strict` argument.
See more:
https://docs.python.org/library/ipaddress.html#ipaddress.IPv4Network
"""
if isinstance(input_value, IPv4Network):
return input_value
try:
return IPv4Network(input_value)
except ValueError:
raise PydanticCustomError('ip_v4_network', 'Input is not a valid IPv4 network')
def ip_v6_network_validator(input_value: Any, /) -> IPv6Network:
"""Assume IPv6Network initialised with a default `strict` argument.
See more:
https://docs.python.org/library/ipaddress.html#ipaddress.IPv6Network
"""
if isinstance(input_value, IPv6Network):
return input_value
try:
return IPv6Network(input_value)
except ValueError:
raise PydanticCustomError('ip_v6_network', 'Input is not a valid IPv6 network')
def ip_v4_interface_validator(input_value: Any, /) -> IPv4Interface:
if isinstance(input_value, IPv4Interface):
return input_value
try:
return IPv4Interface(input_value)
except ValueError:
raise PydanticCustomError('ip_v4_interface', 'Input is not a valid IPv4 interface')
def ip_v6_interface_validator(input_value: Any, /) -> IPv6Interface:
if isinstance(input_value, IPv6Interface):
return input_value
try:
return IPv6Interface(input_value)
except ValueError:
raise PydanticCustomError('ip_v6_interface', 'Input is not a valid IPv6 interface')
def fraction_validator(input_value: Any, /) -> Fraction:
if isinstance(input_value, Fraction):
return input_value
try:
return Fraction(input_value)
except ValueError:
raise PydanticCustomError('fraction_parsing', 'Input is not a valid fraction')
def forbid_inf_nan_check(x: Any) -> Any:
if not math.isfinite(x):
raise PydanticKnownError('finite_number')
return x
def _safe_repr(v: Any) -> int | float | str:
"""The context argument for `PydanticKnownError` requires a number or str type, so we do a simple repr() coercion for types like timedelta.
See tests/test_types.py::test_annotated_metadata_any_order for some context.
"""
if isinstance(v, (int, float, str)):
return v
return repr(v)
def greater_than_validator(x: Any, gt: Any) -> Any:
try:
if not (x > gt):
raise PydanticKnownError('greater_than', {'gt': _safe_repr(gt)})
return x
except TypeError:
raise TypeError(f"Unable to apply constraint 'gt' to supplied value {x}")
def greater_than_or_equal_validator(x: Any, ge: Any) -> Any:
try:
if not (x >= ge):
raise PydanticKnownError('greater_than_equal', {'ge': _safe_repr(ge)})
return x
except TypeError:
raise TypeError(f"Unable to apply constraint 'ge' to supplied value {x}")
def less_than_validator(x: Any, lt: Any) -> Any:
try:
if not (x < lt):
raise PydanticKnownError('less_than', {'lt': _safe_repr(lt)})
return x
except TypeError:
raise TypeError(f"Unable to apply constraint 'lt' to supplied value {x}")
def less_than_or_equal_validator(x: Any, le: Any) -> Any:
try:
if not (x <= le):
raise PydanticKnownError('less_than_equal', {'le': _safe_repr(le)})
return x
except TypeError:
raise TypeError(f"Unable to apply constraint 'le' to supplied value {x}")
def multiple_of_validator(x: Any, multiple_of: Any) -> Any:
try:
if x % multiple_of:
raise PydanticKnownError('multiple_of', {'multiple_of': _safe_repr(multiple_of)})
return x
except TypeError:
raise TypeError(f"Unable to apply constraint 'multiple_of' to supplied value {x}")
def min_length_validator(x: Any, min_length: Any) -> Any:
try:
if not (len(x) >= min_length):
raise PydanticKnownError(
'too_short', {'field_type': 'Value', 'min_length': min_length, 'actual_length': len(x)}
)
return x
except TypeError:
raise TypeError(f"Unable to apply constraint 'min_length' to supplied value {x}")
def max_length_validator(x: Any, max_length: Any) -> Any:
try:
if len(x) > max_length:
raise PydanticKnownError(
'too_long',
{'field_type': 'Value', 'max_length': max_length, 'actual_length': len(x)},
)
return x
except TypeError:
raise TypeError(f"Unable to apply constraint 'max_length' to supplied value {x}")
def _extract_decimal_digits_info(decimal: Decimal) -> tuple[int, int]:
"""Compute the total number of digits and decimal places for a given [`Decimal`][decimal.Decimal] instance.
This function handles both normalized and non-normalized Decimal instances.
Example: Decimal('1.230') -> 4 digits, 3 decimal places
Args:
decimal (Decimal): The decimal number to analyze.
Returns:
tuple[int, int]: A tuple containing the number of decimal places and total digits.
Though this could be divided into two separate functions, the logic is easier to follow if we couple the computation
of the number of decimals and digits together.
"""
try:
decimal_tuple = decimal.as_tuple()
assert isinstance(decimal_tuple.exponent, int)
exponent = decimal_tuple.exponent
num_digits = len(decimal_tuple.digits)
if exponent >= 0:
# A positive exponent adds that many trailing zeros
# Ex: digit_tuple=(1, 2, 3), exponent=2 -> 12300 -> 0 decimal places, 5 digits
num_digits += exponent
decimal_places = 0
else:
# If the absolute value of the negative exponent is larger than the
# number of digits, then it's the same as the number of digits,
# because it'll consume all the digits in digit_tuple and then
# add abs(exponent) - len(digit_tuple) leading zeros after the decimal point.
# Ex: digit_tuple=(1, 2, 3), exponent=-2 -> 1.23 -> 2 decimal places, 3 digits
# Ex: digit_tuple=(1, 2, 3), exponent=-4 -> 0.0123 -> 4 decimal places, 4 digits
decimal_places = abs(exponent)
num_digits = max(num_digits, decimal_places)
return decimal_places, num_digits
except (AssertionError, AttributeError):
raise TypeError(f'Unable to extract decimal digits info from supplied value {decimal}')
def max_digits_validator(x: Any, max_digits: Any) -> Any:
try:
_, num_digits = _extract_decimal_digits_info(x)
_, normalized_num_digits = _extract_decimal_digits_info(x.normalize())
if (num_digits > max_digits) and (normalized_num_digits > max_digits):
raise PydanticKnownError(
'decimal_max_digits',
{'max_digits': max_digits},
)
return x
except TypeError:
raise TypeError(f"Unable to apply constraint 'max_digits' to supplied value {x}")
def decimal_places_validator(x: Any, decimal_places: Any) -> Any:
try:
decimal_places_, _ = _extract_decimal_digits_info(x)
if decimal_places_ > decimal_places:
normalized_decimal_places, _ = _extract_decimal_digits_info(x.normalize())
if normalized_decimal_places > decimal_places:
raise PydanticKnownError(
'decimal_max_places',
{'decimal_places': decimal_places},
)
return x
except TypeError:
raise TypeError(f"Unable to apply constraint 'decimal_places' to supplied value {x}")
def deque_validator(input_value: Any, handler: core_schema.ValidatorFunctionWrapHandler) -> collections.deque[Any]:
return collections.deque(handler(input_value), maxlen=getattr(input_value, 'maxlen', None))
def defaultdict_validator(
input_value: Any, handler: core_schema.ValidatorFunctionWrapHandler, default_default_factory: Callable[[], Any]
) -> collections.defaultdict[Any, Any]:
if isinstance(input_value, collections.defaultdict):
default_factory = input_value.default_factory
return collections.defaultdict(default_factory, handler(input_value))
else:
return collections.defaultdict(default_default_factory, handler(input_value))
def get_defaultdict_default_default_factory(values_source_type: Any) -> Callable[[], Any]:
FieldInfo = import_cached_field_info()
values_type_origin = get_origin(values_source_type)
def infer_default() -> Callable[[], Any]:
allowed_default_types: dict[Any, Any] = {
tuple: tuple,
collections.abc.Sequence: tuple,
collections.abc.MutableSequence: list,
list: list,
typing.Sequence: list,
set: set,
typing.MutableSet: set,
collections.abc.MutableSet: set,
collections.abc.Set: frozenset,
typing.MutableMapping: dict,
typing.Mapping: dict,
collections.abc.Mapping: dict,
collections.abc.MutableMapping: dict,
float: float,
int: int,
str: str,
bool: bool,
}
values_type = values_type_origin or values_source_type
instructions = 'set using `DefaultDict[..., Annotated[..., Field(default_factory=...)]]`'
if typing_objects.is_typevar(values_type):
def type_var_default_factory() -> None:
raise RuntimeError(
'Generic defaultdict cannot be used without a concrete value type or an'
' explicit default factory, ' + instructions
)
return type_var_default_factory
elif values_type not in allowed_default_types:
# a somewhat subjective set of types that have reasonable default values
allowed_msg = ', '.join([t.__name__ for t in set(allowed_default_types.values())])
raise PydanticSchemaGenerationError(
f'Unable to infer a default factory for keys of type {values_source_type}.'
f' Only {allowed_msg} are supported, other types require an explicit default factory'
' ' + instructions
)
return allowed_default_types[values_type]
# Assume Annotated[..., Field(...)]
if typing_objects.is_annotated(values_type_origin):
field_info = next((v for v in get_args(values_source_type) if isinstance(v, FieldInfo)), None)
else:
field_info = None
if field_info and field_info.default_factory:
# Assume the default factory does not take any argument:
default_default_factory = cast(Callable[[], Any], field_info.default_factory)
else:
default_default_factory = infer_default()
return default_default_factory
def validate_str_is_valid_iana_tz(value: Any, /) -> ZoneInfo:
if isinstance(value, ZoneInfo):
return value
try:
return ZoneInfo(value)
except (ZoneInfoNotFoundError, ValueError, TypeError):
raise PydanticCustomError('zoneinfo_str', 'invalid timezone: {value}', {'value': value})
NUMERIC_VALIDATOR_LOOKUP: dict[str, Callable] = {
'gt': greater_than_validator,
'ge': greater_than_or_equal_validator,
'lt': less_than_validator,
'le': less_than_or_equal_validator,
'multiple_of': multiple_of_validator,
'min_length': min_length_validator,
'max_length': max_length_validator,
'max_digits': max_digits_validator,
'decimal_places': decimal_places_validator,
}
IpType = Union[IPv4Address, IPv6Address, IPv4Network, IPv6Network, IPv4Interface, IPv6Interface]
IP_VALIDATOR_LOOKUP: dict[type[IpType], Callable] = {
IPv4Address: ip_v4_address_validator,
IPv6Address: ip_v6_address_validator,
IPv4Network: ip_v4_network_validator,
IPv6Network: ip_v6_network_validator,
IPv4Interface: ip_v4_interface_validator,
IPv6Interface: ip_v6_interface_validator,
}
MAPPING_ORIGIN_MAP: dict[Any, Any] = {
typing.DefaultDict: collections.defaultdict, # noqa: UP006
collections.defaultdict: collections.defaultdict,
typing.OrderedDict: collections.OrderedDict, # noqa: UP006
collections.OrderedDict: collections.OrderedDict,
typing_extensions.OrderedDict: collections.OrderedDict,
typing.Counter: collections.Counter,
collections.Counter: collections.Counter,
# this doesn't handle subclasses of these
typing.Mapping: dict,
typing.MutableMapping: dict,
# parametrized typing.{Mutable}Mapping creates one of these
collections.abc.Mapping: dict,
collections.abc.MutableMapping: dict,
}