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

This commit is contained in:
2026-07-02 20:01:12 +00:00
parent 211715785a
commit c487f1f189
5 changed files with 703 additions and 0 deletions

View File

@@ -0,0 +1,124 @@
"""Tools to provide pretty/human-readable display of objects."""
from __future__ import annotations as _annotations
import types
from collections.abc import Callable, Collection, Generator, Iterable
from typing import TYPE_CHECKING, Any, ForwardRef, cast
import typing_extensions
from typing_extensions import TypeAlias
from typing_inspection import typing_objects
from typing_inspection.introspection import is_union_origin
from . import _typing_extra
if TYPE_CHECKING:
# TODO remove type error comments when we drop support for Python 3.9
ReprArgs: TypeAlias = Iterable[tuple[str | None, Any]] # pyright: ignore[reportGeneralTypeIssues]
RichReprResult: TypeAlias = Iterable[Any | tuple[Any] | tuple[str, Any] | tuple[str, Any, Any]] # pyright: ignore[reportGeneralTypeIssues]
class PlainRepr(str):
"""String class where repr doesn't include quotes. Useful with Representation when you want to return a string
representation of something that is valid (or pseudo-valid) python.
"""
def __repr__(self) -> str:
return str(self)
class Representation:
# Mixin to provide `__str__`, `__repr__`, and `__pretty__` and `__rich_repr__` methods.
# `__pretty__` is used by [devtools](https://python-devtools.helpmanual.io/).
# `__rich_repr__` is used by [rich](https://rich.readthedocs.io/en/stable/pretty.html).
# (this is not a docstring to avoid adding a docstring to classes which inherit from Representation)
__slots__ = ()
def __repr_args__(self) -> ReprArgs:
"""Returns the attributes to show in __str__, __repr__, and __pretty__ this is generally overridden.
Can either return:
* name - value pairs, e.g.: `[('foo_name', 'foo'), ('bar_name', ['b', 'a', 'r'])]`
* or, just values, e.g.: `[(None, 'foo'), (None, ['b', 'a', 'r'])]`
"""
attrs_names = cast(Collection[str], self.__slots__)
if not attrs_names and hasattr(self, '__dict__'):
attrs_names = self.__dict__.keys()
attrs = ((s, getattr(self, s)) for s in attrs_names)
return [(a, v if v is not self else self.__repr_recursion__(v)) for a, v in attrs if v is not None]
def __repr_name__(self) -> str:
"""Name of the instance's class, used in __repr__."""
return self.__class__.__name__
def __repr_recursion__(self, object: Any) -> str:
"""Returns the string representation of a recursive object."""
# This is copied over from the stdlib `pprint` module:
return f'<Recursion on {type(object).__name__} with id={id(object)}>'
def __repr_str__(self, join_str: str) -> str:
return join_str.join(repr(v) if a is None else f'{a}={v!r}' for a, v in self.__repr_args__())
def __pretty__(self, fmt: Callable[[Any], Any], **kwargs: Any) -> Generator[Any]:
"""Used by devtools (https://python-devtools.helpmanual.io/) to pretty print objects."""
yield self.__repr_name__() + '('
yield 1
for name, value in self.__repr_args__():
if name is not None:
yield name + '='
yield fmt(value)
yield ','
yield 0
yield -1
yield ')'
def __rich_repr__(self) -> RichReprResult:
"""Used by Rich (https://rich.readthedocs.io/en/stable/pretty.html) to pretty print objects."""
for name, field_repr in self.__repr_args__():
if name is None:
yield field_repr
else:
yield name, field_repr
def __str__(self) -> str:
return self.__repr_str__(' ')
def __repr__(self) -> str:
return f'{self.__repr_name__()}({self.__repr_str__(", ")})'
def display_as_type(obj: Any) -> str:
"""Pretty representation of a type, should be as close as possible to the original type definition string.
Takes some logic from `typing._type_repr`.
"""
if isinstance(obj, (types.FunctionType, types.BuiltinFunctionType)):
return obj.__name__
elif obj is ...:
return '...'
elif isinstance(obj, Representation):
return repr(obj)
elif isinstance(obj, ForwardRef) or typing_objects.is_typealiastype(obj):
return str(obj)
if not isinstance(obj, (_typing_extra.typing_base, _typing_extra.WithArgsTypes, type)):
obj = obj.__class__
if is_union_origin(typing_extensions.get_origin(obj)):
args = ', '.join(map(display_as_type, typing_extensions.get_args(obj)))
return f'Union[{args}]'
elif isinstance(obj, _typing_extra.WithArgsTypes):
if typing_objects.is_literal(typing_extensions.get_origin(obj)):
args = ', '.join(map(repr, typing_extensions.get_args(obj)))
else:
args = ', '.join(map(display_as_type, typing_extensions.get_args(obj)))
try:
return f'{obj.__qualname__}[{args}]'
except AttributeError:
return str(obj).replace('typing.', '').replace('typing_extensions.', '') # handles TypeAliasType in 3.12
elif isinstance(obj, type):
return obj.__qualname__
else:
return repr(obj).replace('typing.', '').replace('typing_extensions.', '')

View File

@@ -0,0 +1,212 @@
# pyright: reportTypedDictNotRequiredAccess=false, reportGeneralTypeIssues=false, reportArgumentType=false, reportAttributeAccessIssue=false
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TypedDict
from pydantic_core.core_schema import (
ComputedField,
CoreSchema,
DefinitionReferenceSchema,
SerSchema,
iter_union_choices,
)
from typing_extensions import TypeAlias
AllSchemas: TypeAlias = 'CoreSchema | SerSchema | ComputedField'
class GatherResult(TypedDict):
"""Schema traversing result."""
collected_references: dict[str, DefinitionReferenceSchema | None]
"""The collected definition references.
If a definition reference schema can be inlined, it means that there is
only one in the whole core schema. As such, it is stored as the value.
Otherwise, the value is set to `None`.
"""
deferred_discriminator_schemas: list[CoreSchema]
"""The list of core schemas having the discriminator application deferred."""
class MissingDefinitionError(LookupError):
"""A reference was pointing to a non-existing core schema."""
def __init__(self, schema_reference: str, /) -> None:
self.schema_reference = schema_reference
@dataclass
class GatherContext:
"""The current context used during core schema traversing.
Context instances should only be used during schema traversing.
"""
definitions: dict[str, CoreSchema]
"""The available definitions."""
deferred_discriminator_schemas: list[CoreSchema] = field(init=False, default_factory=list)
"""The list of core schemas having the discriminator application deferred.
Internally, these core schemas have a specific key set in the core metadata dict.
"""
collected_references: dict[str, DefinitionReferenceSchema | None] = field(init=False, default_factory=dict)
"""The collected definition references.
If a definition reference schema can be inlined, it means that there is
only one in the whole core schema. As such, it is stored as the value.
Otherwise, the value is set to `None`.
During schema traversing, definition reference schemas can be added as candidates, or removed
(by setting the value to `None`).
"""
def traverse_metadata(schema: AllSchemas, ctx: GatherContext) -> None:
meta = schema.get('metadata')
if meta is not None and 'pydantic_internal_union_discriminator' in meta:
ctx.deferred_discriminator_schemas.append(schema) # pyright: ignore[reportArgumentType]
def traverse_definition_ref(def_ref_schema: DefinitionReferenceSchema, ctx: GatherContext) -> None:
schema_ref = def_ref_schema['schema_ref']
if schema_ref not in ctx.collected_references:
definition = ctx.definitions.get(schema_ref)
if definition is None:
raise MissingDefinitionError(schema_ref)
# The `'definition-ref'` schema was only encountered once, make it
# a candidate to be inlined:
ctx.collected_references[schema_ref] = def_ref_schema
traverse_schema(definition, ctx)
if 'serialization' in def_ref_schema:
traverse_schema(def_ref_schema['serialization'], ctx)
traverse_metadata(def_ref_schema, ctx)
else:
# The `'definition-ref'` schema was already encountered, meaning
# the previously encountered schema (and this one) can't be inlined:
ctx.collected_references[schema_ref] = None
def traverse_schema(schema: AllSchemas, context: GatherContext) -> None:
# TODO When we drop 3.9, use a match statement to get better type checking and remove
# file-level type ignore.
# (the `'type'` could also be fetched in every `if/elif` statement, but this alters performance).
schema_type = schema['type']
if schema_type == 'definition-ref':
traverse_definition_ref(schema, context)
# `traverse_definition_ref` handles the possible serialization and metadata schemas:
return
elif schema_type == 'definitions':
traverse_schema(schema['schema'], context)
for definition in schema['definitions']:
traverse_schema(definition, context)
elif schema_type in {'list', 'set', 'frozenset', 'generator'}:
if 'items_schema' in schema:
traverse_schema(schema['items_schema'], context)
elif schema_type == 'tuple':
if 'items_schema' in schema:
for s in schema['items_schema']:
traverse_schema(s, context)
elif schema_type == 'dict':
if 'keys_schema' in schema:
traverse_schema(schema['keys_schema'], context)
if 'values_schema' in schema:
traverse_schema(schema['values_schema'], context)
elif schema_type == 'union':
for choice in iter_union_choices(schema):
traverse_schema(choice, context)
elif schema_type == 'tagged-union':
for v in schema['choices'].values():
traverse_schema(v, context)
elif schema_type == 'chain':
for step in schema['steps']:
traverse_schema(step, context)
elif schema_type == 'lax-or-strict':
traverse_schema(schema['lax_schema'], context)
traverse_schema(schema['strict_schema'], context)
elif schema_type == 'json-or-python':
traverse_schema(schema['json_schema'], context)
traverse_schema(schema['python_schema'], context)
elif schema_type in {'model-fields', 'typed-dict'}:
if 'extras_schema' in schema:
traverse_schema(schema['extras_schema'], context)
if 'computed_fields' in schema:
for s in schema['computed_fields']:
traverse_schema(s, context)
for s in schema['fields'].values():
traverse_schema(s, context)
elif schema_type == 'dataclass-args':
if 'computed_fields' in schema:
for s in schema['computed_fields']:
traverse_schema(s, context)
for s in schema['fields']:
traverse_schema(s, context)
elif schema_type == 'arguments':
for s in schema['arguments_schema']:
traverse_schema(s['schema'], context)
if 'var_args_schema' in schema:
traverse_schema(schema['var_args_schema'], context)
if 'var_kwargs_schema' in schema:
traverse_schema(schema['var_kwargs_schema'], context)
elif schema_type == 'arguments-v3':
for s in schema['arguments_schema']:
traverse_schema(s['schema'], context)
elif schema_type == 'call':
traverse_schema(schema['arguments_schema'], context)
if 'return_schema' in schema:
traverse_schema(schema['return_schema'], context)
elif schema_type == 'computed-field':
traverse_schema(schema['return_schema'], context)
elif schema_type == 'function-before':
if 'schema' in schema:
traverse_schema(schema['schema'], context)
if 'json_schema_input_schema' in schema:
traverse_schema(schema['json_schema_input_schema'], context)
elif schema_type == 'function-plain':
# TODO duplicate schema types for serializers and validators, needs to be deduplicated.
if 'return_schema' in schema:
traverse_schema(schema['return_schema'], context)
if 'json_schema_input_schema' in schema:
traverse_schema(schema['json_schema_input_schema'], context)
elif schema_type == 'function-wrap':
# TODO duplicate schema types for serializers and validators, needs to be deduplicated.
if 'return_schema' in schema:
traverse_schema(schema['return_schema'], context)
if 'schema' in schema:
traverse_schema(schema['schema'], context)
if 'json_schema_input_schema' in schema:
traverse_schema(schema['json_schema_input_schema'], context)
else:
if 'schema' in schema:
traverse_schema(schema['schema'], context)
if 'serialization' in schema:
traverse_schema(schema['serialization'], context)
traverse_metadata(schema, context)
def gather_schemas_for_cleaning(schema: CoreSchema, definitions: dict[str, CoreSchema]) -> GatherResult:
"""Traverse the core schema and definitions and return the necessary information for schema cleaning.
During the core schema traversing, any `'definition-ref'` schema is:
- Validated: the reference must point to an existing definition. If this is not the case, a
`MissingDefinitionError` exception is raised.
- Stored in the context: the actual reference is stored in the context. Depending on whether
the `'definition-ref'` schema is encountered more that once, the schema itself is also
saved in the context to be inlined (i.e. replaced by the definition it points to).
"""
context = GatherContext(definitions)
traverse_schema(schema, context)
return {
'collected_references': context.collected_references,
'deferred_discriminator_schemas': context.deferred_discriminator_schemas,
}

View File

@@ -0,0 +1,125 @@
"""Types and utility functions used by various other internal tools."""
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Callable, Literal
from pydantic_core import core_schema
from ..annotated_handlers import GetCoreSchemaHandler, GetJsonSchemaHandler
if TYPE_CHECKING:
from ..json_schema import GenerateJsonSchema, JsonSchemaValue
from ._core_utils import CoreSchemaOrField
from ._generate_schema import GenerateSchema
from ._namespace_utils import NamespacesTuple
GetJsonSchemaFunction = Callable[[CoreSchemaOrField, GetJsonSchemaHandler], JsonSchemaValue]
HandlerOverride = Callable[[CoreSchemaOrField], JsonSchemaValue]
class GenerateJsonSchemaHandler(GetJsonSchemaHandler):
"""JsonSchemaHandler implementation that doesn't do ref unwrapping by default.
This is used for any Annotated metadata so that we don't end up with conflicting
modifications to the definition schema.
Used internally by Pydantic, please do not rely on this implementation.
See `GetJsonSchemaHandler` for the handler API.
"""
def __init__(self, generate_json_schema: GenerateJsonSchema, handler_override: HandlerOverride | None) -> None:
self.generate_json_schema = generate_json_schema
self.handler = handler_override or generate_json_schema.generate_inner
self.mode = generate_json_schema.mode
def __call__(self, core_schema: CoreSchemaOrField, /) -> JsonSchemaValue:
return self.handler(core_schema)
def resolve_ref_schema(self, maybe_ref_json_schema: JsonSchemaValue) -> JsonSchemaValue:
"""Resolves `$ref` in the json schema.
This returns the input json schema if there is no `$ref` in json schema.
Args:
maybe_ref_json_schema: The input json schema that may contains `$ref`.
Returns:
Resolved json schema.
Raises:
LookupError: If it can't find the definition for `$ref`.
"""
if '$ref' not in maybe_ref_json_schema:
return maybe_ref_json_schema
ref = maybe_ref_json_schema['$ref']
json_schema = self.generate_json_schema.get_schema_from_definitions(ref)
if json_schema is None:
raise LookupError(
f'Could not find a ref for {ref}.'
' Maybe you tried to call resolve_ref_schema from within a recursive model?'
)
return json_schema
class CallbackGetCoreSchemaHandler(GetCoreSchemaHandler):
"""Wrapper to use an arbitrary function as a `GetCoreSchemaHandler`.
Used internally by Pydantic, please do not rely on this implementation.
See `GetCoreSchemaHandler` for the handler API.
"""
def __init__(
self,
handler: Callable[[Any], core_schema.CoreSchema],
generate_schema: GenerateSchema,
ref_mode: Literal['to-def', 'unpack'] = 'to-def',
) -> None:
self._handler = handler
self._generate_schema = generate_schema
self._ref_mode = ref_mode
def __call__(self, source_type: Any, /) -> core_schema.CoreSchema:
schema = self._handler(source_type)
if self._ref_mode == 'to-def':
ref = schema.get('ref')
if ref is not None:
return self._generate_schema.defs.create_definition_reference_schema(schema)
return schema
else: # ref_mode = 'unpack'
return self.resolve_ref_schema(schema)
def _get_types_namespace(self) -> NamespacesTuple:
return self._generate_schema._types_namespace
def generate_schema(self, source_type: Any, /) -> core_schema.CoreSchema:
return self._generate_schema.generate_schema(source_type)
@property
def field_name(self) -> str | None:
return self._generate_schema.field_name_stack.get()
def resolve_ref_schema(self, maybe_ref_schema: core_schema.CoreSchema) -> core_schema.CoreSchema:
"""Resolves reference in the core schema.
Args:
maybe_ref_schema: The input core schema that may contains reference.
Returns:
Resolved core schema.
Raises:
LookupError: If it can't find the definition for reference.
"""
if maybe_ref_schema['type'] == 'definition-ref':
ref = maybe_ref_schema['schema_ref']
definition = self._generate_schema.defs.get_schema_from_ref(ref)
if definition is None:
raise LookupError(
f'Could not find a ref for {ref}.'
' Maybe you tried to call resolve_ref_schema from within a recursive model?'
)
return definition
elif maybe_ref_schema['type'] == 'definitions':
return self.resolve_ref_schema(maybe_ref_schema['schema'])
return maybe_ref_schema

View File

@@ -0,0 +1,53 @@
from __future__ import annotations
import collections
import collections.abc
import typing
from typing import Any
from pydantic_core import PydanticOmit, core_schema
SEQUENCE_ORIGIN_MAP: dict[Any, Any] = {
typing.Deque: collections.deque, # noqa: UP006
collections.deque: collections.deque,
list: list,
typing.List: list, # noqa: UP006
tuple: tuple,
typing.Tuple: tuple, # noqa: UP006
set: set,
typing.AbstractSet: set,
typing.Set: set, # noqa: UP006
frozenset: frozenset,
typing.FrozenSet: frozenset, # noqa: UP006
typing.Sequence: list,
typing.MutableSequence: list,
typing.MutableSet: set,
# this doesn't handle subclasses of these
# parametrized typing.Set creates one of these
collections.abc.MutableSet: set,
collections.abc.Set: frozenset,
}
def serialize_sequence_via_list(
v: Any, handler: core_schema.SerializerFunctionWrapHandler, info: core_schema.SerializationInfo
) -> Any:
items: list[Any] = []
mapped_origin = SEQUENCE_ORIGIN_MAP.get(type(v), None)
if mapped_origin is None:
# we shouldn't hit this branch, should probably add a serialization error or something
return v
for index, item in enumerate(v):
try:
v = handler(item, index)
except PydanticOmit: # noqa: PERF203
pass
else:
items.append(v)
if info.mode_is_json():
return items
else:
return mapped_origin(items)

View File

@@ -0,0 +1,189 @@
from __future__ import annotations
import dataclasses
from inspect import Parameter, Signature
from typing import TYPE_CHECKING, Any, Callable
from pydantic_core import PydanticUndefined
from ._typing_extra import signature_no_eval
from ._utils import is_valid_identifier
if TYPE_CHECKING:
from ..config import ExtraValues
from ..fields import FieldInfo
# Copied over from stdlib dataclasses
class _HAS_DEFAULT_FACTORY_CLASS:
def __repr__(self):
return '<factory>'
_HAS_DEFAULT_FACTORY = _HAS_DEFAULT_FACTORY_CLASS()
def _field_name_for_signature(field_name: str, field_info: FieldInfo) -> str:
"""Extract the correct name to use for the field when generating a signature.
Assuming the field has a valid alias, this will return the alias. Otherwise, it will return the field name.
First priority is given to the alias, then the validation_alias, then the field name.
Args:
field_name: The name of the field
field_info: The corresponding FieldInfo object.
Returns:
The correct name to use when generating a signature.
"""
if isinstance(field_info.alias, str) and is_valid_identifier(field_info.alias):
return field_info.alias
if isinstance(field_info.validation_alias, str) and is_valid_identifier(field_info.validation_alias):
return field_info.validation_alias
return field_name
def _process_param_defaults(param: Parameter) -> Parameter:
"""Modify the signature for a parameter in a dataclass where the default value is a FieldInfo instance.
Args:
param (Parameter): The parameter
Returns:
Parameter: The custom processed parameter
"""
from ..fields import FieldInfo
param_default = param.default
if isinstance(param_default, FieldInfo):
annotation = param.annotation
# Replace the annotation if appropriate
# inspect does "clever" things to show annotations as strings because we have
# `from __future__ import annotations` in main, we don't want that
if annotation == 'Any':
annotation = Any
# Replace the field default
default = param_default.default
if default is PydanticUndefined:
if param_default.default_factory is None:
default = Signature.empty
else:
# this is used by dataclasses to indicate a factory exists:
default = dataclasses._HAS_DEFAULT_FACTORY # type: ignore
return param.replace(
annotation=annotation, name=_field_name_for_signature(param.name, param_default), default=default
)
return param
def _generate_signature_parameters( # noqa: C901 (ignore complexity, could use a refactor)
init: Callable[..., None],
fields: dict[str, FieldInfo],
validate_by_name: bool,
extra: ExtraValues | None,
) -> dict[str, Parameter]:
"""Generate a mapping of parameter names to Parameter objects for a pydantic BaseModel or dataclass."""
from itertools import islice
present_params = signature_no_eval(init).parameters.values()
merged_params: dict[str, Parameter] = {}
var_kw = None
use_var_kw = False
for param in islice(present_params, 1, None): # skip self arg
# inspect does "clever" things to show annotations as strings because we have
# `from __future__ import annotations` in main, we don't want that
if fields.get(param.name):
# exclude params with init=False
if getattr(fields[param.name], 'init', True) is False:
continue
param = param.replace(name=_field_name_for_signature(param.name, fields[param.name]))
if param.annotation == 'Any':
param = param.replace(annotation=Any)
if param.kind is param.VAR_KEYWORD:
var_kw = param
continue
merged_params[param.name] = param
if var_kw: # if custom init has no var_kw, fields which are not declared in it cannot be passed through
allow_names = validate_by_name
for field_name, field in fields.items():
# when alias is a str it should be used for signature generation
param_name = _field_name_for_signature(field_name, field)
if field_name in merged_params or param_name in merged_params:
continue
if not is_valid_identifier(param_name):
if allow_names:
param_name = field_name
else:
use_var_kw = True
continue
if field.is_required():
default = Parameter.empty
elif field.default_factory is not None:
# Mimics stdlib dataclasses:
default = _HAS_DEFAULT_FACTORY
else:
default = field.default
merged_params[param_name] = Parameter(
param_name,
Parameter.KEYWORD_ONLY,
annotation=field.rebuild_annotation(),
default=default,
)
if extra == 'allow':
use_var_kw = True
if var_kw and use_var_kw:
# Make sure the parameter for extra kwargs
# does not have the same name as a field
default_model_signature = [
('self', Parameter.POSITIONAL_ONLY),
('data', Parameter.VAR_KEYWORD),
]
if [(p.name, p.kind) for p in present_params] == default_model_signature:
# if this is the standard model signature, use extra_data as the extra args name
var_kw_name = 'extra_data'
else:
# else start from var_kw
var_kw_name = var_kw.name
# generate a name that's definitely unique
while var_kw_name in fields:
var_kw_name += '_'
merged_params[var_kw_name] = var_kw.replace(name=var_kw_name)
return merged_params
def generate_pydantic_signature(
init: Callable[..., None],
fields: dict[str, FieldInfo],
validate_by_name: bool,
extra: ExtraValues | None,
is_dataclass: bool = False,
) -> Signature:
"""Generate signature for a pydantic BaseModel or dataclass.
Args:
init: The class init.
fields: The model fields.
validate_by_name: The `validate_by_name` value of the config.
extra: The `extra` value of the config.
is_dataclass: Whether the model is a dataclass.
Returns:
The dataclass/BaseModel subclass signature.
"""
merged_params = _generate_signature_parameters(init, fields, validate_by_name, extra)
if is_dataclass:
merged_params = {k: _process_param_defaults(v) for k, v in merged_params.items()}
return Signature(parameters=list(merged_params.values()), return_annotation=None)