Coverage for pydantic/_internal/_generate_schema.py: 95.15%
1307 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-05 10:27 +0000
1"""Convert python types to pydantic-core schema."""
3from __future__ import annotations as _annotations 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
5import collections.abc 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
6import dataclasses 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
7import datetime 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
8import inspect 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
9import os 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
10import pathlib 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
11import re 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
12import sys 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
13import typing 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
14import warnings 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
15from collections.abc import Generator, Iterable, Iterator, Mapping 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
16from contextlib import contextmanager 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
17from copy import copy 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
18from decimal import Decimal 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
19from enum import Enum 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
20from fractions import Fraction 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
21from functools import partial 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
22from inspect import Parameter, _ParameterKind, signature 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
23from ipaddress import IPv4Address, IPv4Interface, IPv4Network, IPv6Address, IPv6Interface, IPv6Network 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
24from itertools import chain 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
25from operator import attrgetter 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
26from types import FunctionType, GenericAlias, LambdaType, MethodType 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
27from typing import ( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
28 TYPE_CHECKING,
29 Any,
30 Callable,
31 Final,
32 ForwardRef,
33 Literal,
34 TypeVar,
35 Union,
36 cast,
37 overload,
38)
39from uuid import UUID 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
40from zoneinfo import ZoneInfo 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
42import typing_extensions 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
43from pydantic_core import ( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
44 CoreSchema,
45 MultiHostUrl,
46 PydanticCustomError,
47 PydanticSerializationUnexpectedValue,
48 PydanticUndefined,
49 Url,
50 core_schema,
51 to_jsonable_python,
52)
53from typing_extensions import TypeAlias, TypeAliasType, TypedDict, get_args, get_origin, is_typeddict 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
54from typing_inspection import typing_objects 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
55from typing_inspection.introspection import AnnotationSource, get_literal_values, is_union_origin 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
57from ..aliases import AliasChoices, AliasPath 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
58from ..annotated_handlers import GetCoreSchemaHandler, GetJsonSchemaHandler 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
59from ..config import ConfigDict, JsonDict, JsonEncoder, JsonSchemaExtraCallable 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
60from ..errors import PydanticSchemaGenerationError, PydanticUndefinedAnnotation, PydanticUserError 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
61from ..functional_validators import AfterValidator, BeforeValidator, FieldValidatorModes, PlainValidator, WrapValidator 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
62from ..json_schema import JsonSchemaValue 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
63from ..version import version_short 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
64from ..warnings import ArbitraryTypeWarning, PydanticDeprecatedSince20, UnsupportedFieldAttributeWarning 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
65from . import _decorators, _discriminated_union, _known_annotated_metadata, _repr, _typing_extra 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
66from ._config import ConfigWrapper, ConfigWrapperStack 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
67from ._core_metadata import CoreMetadata, update_core_metadata 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
68from ._core_utils import ( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
69 get_ref,
70 get_type_ref,
71 is_list_like_schema_with_items_schema,
72)
73from ._decorators import ( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
74 Decorator,
75 DecoratorInfos,
76 FieldSerializerDecoratorInfo,
77 FieldValidatorDecoratorInfo,
78 ModelSerializerDecoratorInfo,
79 ModelValidatorDecoratorInfo,
80 RootValidatorDecoratorInfo,
81 ValidatorDecoratorInfo,
82 get_attribute_from_bases,
83 inspect_field_serializer,
84 inspect_model_serializer,
85 inspect_validator,
86)
87from ._docs_extraction import extract_docstrings_from_cls 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
88from ._fields import ( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
89 collect_dataclass_fields,
90 rebuild_dataclass_fields,
91 rebuild_model_fields,
92 takes_validated_data_argument,
93 update_field_from_config,
94)
95from ._forward_ref import PydanticRecursiveRef 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
96from ._generics import get_standard_typevars_map, replace_types 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
97from ._import_utils import import_cached_base_model, import_cached_field_info 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
98from ._mock_val_ser import MockCoreSchema 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
99from ._namespace_utils import NamespacesTuple, NsResolver 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
100from ._schema_gather import MissingDefinitionError, gather_schemas_for_cleaning 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
101from ._schema_generation_shared import CallbackGetCoreSchemaHandler 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
102from ._utils import lenient_issubclass, smart_deepcopy 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
104if TYPE_CHECKING: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
105 from ..fields import ComputedFieldInfo, FieldInfo
106 from ..main import BaseModel
107 from ..types import Discriminator
108 from ._dataclasses import StandardDataclass
109 from ._schema_generation_shared import GetJsonSchemaFunction
111_SUPPORTS_TYPEDDICT = sys.version_info >= (3, 12) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
113FieldDecoratorInfo = Union[ValidatorDecoratorInfo, FieldValidatorDecoratorInfo, FieldSerializerDecoratorInfo] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
114FieldDecoratorInfoType = TypeVar('FieldDecoratorInfoType', bound=FieldDecoratorInfo) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
115AnyFieldDecorator = Union[ 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
116 Decorator[ValidatorDecoratorInfo],
117 Decorator[FieldValidatorDecoratorInfo],
118 Decorator[FieldSerializerDecoratorInfo],
119]
121ModifyCoreSchemaWrapHandler: TypeAlias = GetCoreSchemaHandler 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
122GetCoreSchemaFunction: TypeAlias = Callable[[Any, ModifyCoreSchemaWrapHandler], core_schema.CoreSchema] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
123ParametersCallback: TypeAlias = "Callable[[int, str, Any], Literal['skip'] | None]" 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
125TUPLE_TYPES: list[type] = [typing.Tuple, tuple] # noqa: UP006 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
126LIST_TYPES: list[type] = [typing.List, list, collections.abc.MutableSequence] # noqa: UP006 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
127SET_TYPES: list[type] = [typing.Set, set, collections.abc.MutableSet] # noqa: UP006 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
128FROZEN_SET_TYPES: list[type] = [typing.FrozenSet, frozenset, collections.abc.Set] # noqa: UP006 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
129DICT_TYPES: list[type] = [typing.Dict, dict] # noqa: UP006 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
130IP_TYPES: list[type] = [IPv4Address, IPv4Interface, IPv4Network, IPv6Address, IPv6Interface, IPv6Network] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
131SEQUENCE_TYPES: list[type] = [typing.Sequence, collections.abc.Sequence] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
132ITERABLE_TYPES: list[type] = [typing.Iterable, collections.abc.Iterable, typing.Generator, collections.abc.Generator] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
133TYPE_TYPES: list[type] = [typing.Type, type] # noqa: UP006 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
134PATTERN_TYPES: list[type] = [typing.Pattern, re.Pattern] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
135PATH_TYPES: list[type] = [ 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
136 os.PathLike,
137 pathlib.Path,
138 pathlib.PurePath,
139 pathlib.PosixPath,
140 pathlib.PurePosixPath,
141 pathlib.PureWindowsPath,
142]
143MAPPING_TYPES = [ 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
144 typing.Mapping,
145 typing.MutableMapping,
146 collections.abc.Mapping,
147 collections.abc.MutableMapping,
148 collections.OrderedDict,
149 typing_extensions.OrderedDict,
150 typing.DefaultDict, # noqa: UP006
151 collections.defaultdict,
152]
153COUNTER_TYPES = [collections.Counter, typing.Counter] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
154DEQUE_TYPES: list[type] = [collections.deque, typing.Deque] # noqa: UP006 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
156# Note: This does not play very well with type checkers. For example,
157# `a: LambdaType = lambda x: x` will raise a type error by Pyright.
158ValidateCallSupportedTypes = Union[ 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
159 LambdaType,
160 FunctionType,
161 MethodType,
162 partial,
163]
165VALIDATE_CALL_SUPPORTED_TYPES = get_args(ValidateCallSupportedTypes) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
166UNSUPPORTED_STANDALONE_FIELDINFO_ATTRIBUTES = [ 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
167 ('alias', None),
168 ('validation_alias', None),
169 ('serialization_alias', None),
170 # will be set if any alias is set, so disable it to avoid double warnings:
171 # 'alias_priority',
172 ('default', PydanticUndefined),
173 ('default_factory', None),
174 ('exclude', None),
175 ('deprecated', None),
176 ('repr', True),
177 ('validate_default', None),
178 ('frozen', None),
179 ('init', None),
180 ('init_var', None),
181 ('kw_only', None),
182]
183"""`FieldInfo` attributes (and their default value) that can't be used outside of a model (e.g. in a type adapter or a PEP 695 type alias).""" 1cenpuwACDfghrsxyEFGJdjotvzBHI
185_mode_to_validator: dict[ 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
186 FieldValidatorModes, type[BeforeValidator | AfterValidator | PlainValidator | WrapValidator]
187] = {'before': BeforeValidator, 'after': AfterValidator, 'plain': PlainValidator, 'wrap': WrapValidator}
190def check_validator_fields_against_field_name( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
191 info: FieldDecoratorInfo,
192 field: str,
193) -> bool:
194 """Check if field name is in validator fields.
196 Args:
197 info: The field info.
198 field: The field name to check.
200 Returns:
201 `True` if field name is in validator fields, `False` otherwise.
202 """
203 fields = info.fields 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
204 return '*' in fields or field in fields 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
207def check_decorator_fields_exist(decorators: Iterable[AnyFieldDecorator], fields: Iterable[str]) -> None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
208 """Check if the defined fields in decorators exist in `fields` param.
210 It ignores the check for a decorator if the decorator has `*` as field or `check_fields=False`.
212 Args:
213 decorators: An iterable of decorators.
214 fields: An iterable of fields name.
216 Raises:
217 PydanticUserError: If one of the field names does not exist in `fields` param.
218 """
219 fields = set(fields) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
220 for dec in decorators: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
221 if '*' in dec.info.fields: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
222 continue 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
223 if dec.info.check_fields is False: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
224 continue 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
225 for field in dec.info.fields: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
226 if field not in fields: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
227 raise PydanticUserError( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
228 f'Decorators defined with incorrect fields: {dec.cls_ref}.{dec.cls_var_name}'
229 " (use check_fields=False if you're inheriting from the model and intended this)",
230 code='decorator-missing-field',
231 )
234def filter_field_decorator_info_by_field( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
235 validator_functions: Iterable[Decorator[FieldDecoratorInfoType]], field: str
236) -> list[Decorator[FieldDecoratorInfoType]]:
237 return [dec for dec in validator_functions if check_validator_fields_against_field_name(dec.info, field)] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
240def apply_each_item_validators( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
241 schema: core_schema.CoreSchema,
242 each_item_validators: list[Decorator[ValidatorDecoratorInfo]],
243) -> core_schema.CoreSchema:
244 # This V1 compatibility shim should eventually be removed
246 # fail early if each_item_validators is empty
247 if not each_item_validators: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
248 return schema 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
250 # push down any `each_item=True` validators
251 # note that this won't work for any Annotated types that get wrapped by a function validator
252 # but that's okay because that didn't exist in V1
253 if schema['type'] == 'nullable': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
254 schema['schema'] = apply_each_item_validators(schema['schema'], each_item_validators) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
255 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
256 elif schema['type'] == 'tuple': 256 ↛ 257line 256 didn't jump to line 257 because the condition on line 256 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
257 if (variadic_item_index := schema.get('variadic_item_index')) is not None:
258 schema['items_schema'][variadic_item_index] = apply_validators(
259 schema['items_schema'][variadic_item_index],
260 each_item_validators,
261 )
262 elif is_list_like_schema_with_items_schema(schema): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
263 inner_schema = schema.get('items_schema', core_schema.any_schema()) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
264 schema['items_schema'] = apply_validators(inner_schema, each_item_validators) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
265 elif schema['type'] == 'dict': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
266 inner_schema = schema.get('values_schema', core_schema.any_schema()) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
267 schema['values_schema'] = apply_validators(inner_schema, each_item_validators) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
268 else:
269 raise TypeError( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
270 f'`@validator(..., each_item=True)` cannot be applied to fields with a schema of {schema["type"]}'
271 )
272 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
275def _extract_json_schema_info_from_field_info( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
276 info: FieldInfo | ComputedFieldInfo,
277) -> tuple[JsonDict | None, JsonDict | JsonSchemaExtraCallable | None]:
278 json_schema_updates = { 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
279 'title': info.title,
280 'description': info.description,
281 'deprecated': bool(info.deprecated) or info.deprecated == '' or None,
282 'examples': to_jsonable_python(info.examples),
283 }
284 json_schema_updates = {k: v for k, v in json_schema_updates.items() if v is not None} 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
285 return (json_schema_updates or None, info.json_schema_extra) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
288JsonEncoders = dict[type[Any], JsonEncoder] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
291def _add_custom_serialization_from_json_encoders( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
292 json_encoders: JsonEncoders | None, tp: Any, schema: CoreSchema
293) -> CoreSchema:
294 """Iterate over the json_encoders and add the first matching encoder to the schema.
296 Args:
297 json_encoders: A dictionary of types and their encoder functions.
298 tp: The type to check for a matching encoder.
299 schema: The schema to add the encoder to.
300 """
301 if not json_encoders: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
302 return schema 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
303 if 'serialization' in schema: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
304 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
305 # Check the class type and its superclasses for a matching encoder
306 # Decimal.__class__.__mro__ (and probably other cases) doesn't include Decimal itself
307 # if the type is a GenericAlias (e.g. from list[int]) we need to use __class__ instead of .__mro__
308 for base in (tp, *getattr(tp, '__mro__', tp.__class__.__mro__)[:-1]): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
309 encoder = json_encoders.get(base) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
310 if encoder is None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
311 continue 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
313 warnings.warn( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
314 f'`json_encoders` is deprecated. See https://docs.pydantic.dev/{version_short()}/concepts/serialization/#custom-serializers for alternatives',
315 PydanticDeprecatedSince20,
316 )
318 # TODO: in theory we should check that the schema accepts a serialization key
319 schema['serialization'] = core_schema.plain_serializer_function_ser_schema(encoder, when_used='json') 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
320 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
322 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
325class InvalidSchemaError(Exception): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
326 """The core schema is invalid."""
329class GenerateSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
330 """Generate core schema for a Pydantic model, dataclass and types like `str`, `datetime`, ... ."""
332 __slots__ = ( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
333 '_config_wrapper_stack',
334 '_ns_resolver',
335 '_typevars_map',
336 'field_name_stack',
337 'model_type_stack',
338 'defs',
339 )
341 def __init__( 1abcenpuwACDklghrsxyEFGJimdjotvzBHI
342 self,
343 config_wrapper: ConfigWrapper,
344 ns_resolver: NsResolver | None = None,
345 typevars_map: Mapping[TypeVar, Any] | None = None,
346 ) -> None:
347 # we need a stack for recursing into nested models
348 self._config_wrapper_stack = ConfigWrapperStack(config_wrapper) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
349 self._ns_resolver = ns_resolver or NsResolver() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
350 self._typevars_map = typevars_map 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
351 self.field_name_stack = _FieldNameStack() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
352 self.model_type_stack = _ModelTypeStack() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
353 self.defs = _Definitions() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
355 def __init_subclass__(cls) -> None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
356 super().__init_subclass__() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
357 warnings.warn( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
358 'Subclassing `GenerateSchema` is not supported. The API is highly subject to change in minor versions.',
359 UserWarning,
360 stacklevel=2,
361 )
363 @property 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
364 def _config_wrapper(self) -> ConfigWrapper: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
365 return self._config_wrapper_stack.tail 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
367 @property 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
368 def _types_namespace(self) -> NamespacesTuple: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
369 return self._ns_resolver.types_namespace 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
371 @property 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
372 def _arbitrary_types(self) -> bool: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
373 return self._config_wrapper.arbitrary_types_allowed 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
375 # the following methods can be overridden but should be considered
376 # unstable / private APIs
377 def _list_schema(self, items_type: Any) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
378 return core_schema.list_schema(self.generate_schema(items_type)) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
380 def _dict_schema(self, keys_type: Any, values_type: Any) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
381 return core_schema.dict_schema(self.generate_schema(keys_type), self.generate_schema(values_type)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
383 def _set_schema(self, items_type: Any) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
384 return core_schema.set_schema(self.generate_schema(items_type)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
386 def _frozenset_schema(self, items_type: Any) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
387 return core_schema.frozenset_schema(self.generate_schema(items_type)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
389 def _enum_schema(self, enum_type: type[Enum]) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
390 cases: list[Any] = list(enum_type.__members__.values()) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
392 enum_ref = get_type_ref(enum_type) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
393 description = None if not enum_type.__doc__ else inspect.cleandoc(enum_type.__doc__) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
394 if ( 1abqf
395 description == 'An enumeration.'
396 ): # This is the default value provided by enum.EnumMeta.__new__; don't use it
397 description = None 1abceqfklghimdj
398 js_updates = {'title': enum_type.__name__, 'description': description} 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
399 js_updates = {k: v for k, v in js_updates.items() if v is not None} 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
401 sub_type: Literal['str', 'int', 'float'] | None = None 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
402 if issubclass(enum_type, int): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
403 sub_type = 'int' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
404 value_ser_type: core_schema.SerSchema = core_schema.simple_ser_schema('int') 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
405 elif issubclass(enum_type, str): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
406 # this handles `StrEnum` (3.11 only), and also `Foobar(str, Enum)`
407 sub_type = 'str' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
408 value_ser_type = core_schema.simple_ser_schema('str') 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
409 elif issubclass(enum_type, float): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
410 sub_type = 'float' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
411 value_ser_type = core_schema.simple_ser_schema('float') 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
412 else:
413 # TODO this is an ugly hack, how do we trigger an Any schema for serialization?
414 value_ser_type = core_schema.plain_serializer_function_ser_schema(lambda x: x) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
416 if cases: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
418 def get_json_schema(schema: CoreSchema, handler: GetJsonSchemaHandler) -> JsonSchemaValue: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
419 json_schema = handler(schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
420 original_schema = handler.resolve_ref_schema(json_schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
421 original_schema.update(js_updates) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
422 return json_schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
424 # we don't want to add the missing to the schema if it's the default one
425 default_missing = getattr(enum_type._missing_, '__func__', None) is Enum._missing_.__func__ # pyright: ignore[reportFunctionMemberAccess] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
426 enum_schema = core_schema.enum_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
427 enum_type,
428 cases,
429 sub_type=sub_type,
430 missing=None if default_missing else enum_type._missing_,
431 ref=enum_ref,
432 metadata={'pydantic_js_functions': [get_json_schema]},
433 )
435 if self._config_wrapper.use_enum_values: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
436 enum_schema = core_schema.no_info_after_validator_function( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
437 attrgetter('value'), enum_schema, serialization=value_ser_type
438 )
440 return enum_schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
442 else:
444 def get_json_schema_no_cases(_, handler: GetJsonSchemaHandler) -> JsonSchemaValue: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
445 json_schema = handler(core_schema.enum_schema(enum_type, cases, sub_type=sub_type, ref=enum_ref)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
446 original_schema = handler.resolve_ref_schema(json_schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
447 original_schema.update(js_updates) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
448 return json_schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
450 # Use an isinstance check for enums with no cases.
451 # The most important use case for this is creating TypeVar bounds for generics that should
452 # be restricted to enums. This is more consistent than it might seem at first, since you can only
453 # subclass enum.Enum (or subclasses of enum.Enum) if all parent classes have no cases.
454 # We use the get_json_schema function when an Enum subclass has been declared with no cases
455 # so that we can still generate a valid json schema.
456 return core_schema.is_instance_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
457 enum_type,
458 metadata={'pydantic_js_functions': [get_json_schema_no_cases]},
459 )
461 def _ip_schema(self, tp: Any) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
462 from ._validators import IP_VALIDATOR_LOOKUP, IpType 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
464 ip_type_json_schema_format: dict[type[IpType], str] = { 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
465 IPv4Address: 'ipv4',
466 IPv4Network: 'ipv4network',
467 IPv4Interface: 'ipv4interface',
468 IPv6Address: 'ipv6',
469 IPv6Network: 'ipv6network',
470 IPv6Interface: 'ipv6interface',
471 }
473 def ser_ip(ip: Any, info: core_schema.SerializationInfo) -> str | IpType: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
474 if not isinstance(ip, (tp, str)): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
475 raise PydanticSerializationUnexpectedValue( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
476 f"Expected `{tp}` but got `{type(ip)}` with value `'{ip}'` - serialized value may not be as expected."
477 )
478 if info.mode == 'python': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
479 return ip 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
480 return str(ip) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
482 return core_schema.lax_or_strict_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
483 lax_schema=core_schema.no_info_plain_validator_function(IP_VALIDATOR_LOOKUP[tp]),
484 strict_schema=core_schema.json_or_python_schema(
485 json_schema=core_schema.no_info_after_validator_function(tp, core_schema.str_schema()),
486 python_schema=core_schema.is_instance_schema(tp),
487 ),
488 serialization=core_schema.plain_serializer_function_ser_schema(ser_ip, info_arg=True, when_used='always'),
489 metadata={
490 'pydantic_js_functions': [lambda _1, _2: {'type': 'string', 'format': ip_type_json_schema_format[tp]}]
491 },
492 )
494 def _path_schema(self, tp: Any, path_type: Any) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
495 if tp is os.PathLike and (path_type not in {str, bytes} and not typing_objects.is_any(path_type)): 495 ↛ 496line 495 didn't jump to line 496 because the condition on line 495 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
496 raise PydanticUserError(
497 '`os.PathLike` can only be used with `str`, `bytes` or `Any`', code='schema-for-unknown-type'
498 )
500 path_constructor = pathlib.PurePath if tp is os.PathLike else tp 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
501 strict_inner_schema = ( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
502 core_schema.bytes_schema(strict=True) if (path_type is bytes) else core_schema.str_schema(strict=True)
503 )
504 lax_inner_schema = core_schema.bytes_schema() if (path_type is bytes) else core_schema.str_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
506 def path_validator(input_value: str | bytes) -> os.PathLike[Any]: # type: ignore 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
507 try: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
508 if path_type is bytes: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
509 if isinstance(input_value, bytes): 509 ↛ 515line 509 didn't jump to line 515 because the condition on line 509 was always true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
510 try: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
511 input_value = input_value.decode() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
512 except UnicodeDecodeError as e:
513 raise PydanticCustomError('bytes_type', 'Input must be valid bytes') from e
514 else:
515 raise PydanticCustomError('bytes_type', 'Input must be bytes')
516 elif not isinstance(input_value, str): 516 ↛ 517line 516 didn't jump to line 517 because the condition on line 516 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
517 raise PydanticCustomError('path_type', 'Input is not a valid path')
519 return path_constructor(input_value) # type: ignore 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
520 except TypeError as e:
521 raise PydanticCustomError('path_type', 'Input is not a valid path') from e
523 def ser_path(path: Any, info: core_schema.SerializationInfo) -> str | os.PathLike[Any]: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
524 if not isinstance(path, (tp, str)): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
525 raise PydanticSerializationUnexpectedValue( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
526 f"Expected `{tp}` but got `{type(path)}` with value `'{path}'` - serialized value may not be as expected."
527 )
528 if info.mode == 'python': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
529 return path 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
530 return str(path) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
532 instance_schema = core_schema.json_or_python_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
533 json_schema=core_schema.no_info_after_validator_function(path_validator, lax_inner_schema),
534 python_schema=core_schema.is_instance_schema(tp),
535 )
537 schema = core_schema.lax_or_strict_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
538 lax_schema=core_schema.union_schema(
539 [
540 instance_schema,
541 core_schema.no_info_after_validator_function(path_validator, strict_inner_schema),
542 ],
543 custom_error_type='path_type',
544 custom_error_message=f'Input is not a valid path for {tp}',
545 ),
546 strict_schema=instance_schema,
547 serialization=core_schema.plain_serializer_function_ser_schema(ser_path, info_arg=True, when_used='always'),
548 metadata={'pydantic_js_functions': [lambda source, handler: {**handler(source), 'format': 'path'}]},
549 )
550 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
552 def _deque_schema(self, items_type: Any) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
553 from ._serializers import serialize_sequence_via_list 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
554 from ._validators import deque_validator 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
556 item_type_schema = self.generate_schema(items_type) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
558 # we have to use a lax list schema here, because we need to validate the deque's
559 # items via a list schema, but it's ok if the deque itself is not a list
560 list_schema = core_schema.list_schema(item_type_schema, strict=False) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
562 check_instance = core_schema.json_or_python_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
563 json_schema=list_schema,
564 python_schema=core_schema.is_instance_schema(collections.deque, cls_repr='Deque'),
565 )
567 lax_schema = core_schema.no_info_wrap_validator_function(deque_validator, list_schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
569 return core_schema.lax_or_strict_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
570 lax_schema=lax_schema,
571 strict_schema=core_schema.chain_schema([check_instance, lax_schema]),
572 serialization=core_schema.wrap_serializer_function_ser_schema(
573 serialize_sequence_via_list, schema=item_type_schema, info_arg=True
574 ),
575 )
577 def _mapping_schema(self, tp: Any, keys_type: Any, values_type: Any) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
578 from ._validators import MAPPING_ORIGIN_MAP, defaultdict_validator, get_defaultdict_default_default_factory 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
580 mapped_origin = MAPPING_ORIGIN_MAP[tp] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
581 keys_schema = self.generate_schema(keys_type) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
582 with warnings.catch_warnings(): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
583 # We kind of abused `Field()` default factories to be able to specify
584 # the `defaultdict`'s `default_factory`. As a consequence, we get warnings
585 # as normally `FieldInfo.default_factory` is unsupported in the context where
586 # `Field()` is used and our only solution is to ignore them (note that this might
587 # wrongfully ignore valid warnings, e.g. if the `value_type` is a PEP 695 type alias
588 # with unsupported metadata).
589 warnings.simplefilter('ignore', category=UnsupportedFieldAttributeWarning) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
590 values_schema = self.generate_schema(values_type) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
591 dict_schema = core_schema.dict_schema(keys_schema, values_schema, strict=False) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
593 if mapped_origin is dict: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
594 schema = dict_schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
595 else:
596 check_instance = core_schema.json_or_python_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
597 json_schema=dict_schema,
598 python_schema=core_schema.is_instance_schema(mapped_origin),
599 )
601 if tp is collections.defaultdict: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
602 default_default_factory = get_defaultdict_default_default_factory(values_type) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
603 coerce_instance_wrap = partial( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
604 core_schema.no_info_wrap_validator_function,
605 partial(defaultdict_validator, default_default_factory=default_default_factory),
606 )
607 else:
608 coerce_instance_wrap = partial(core_schema.no_info_after_validator_function, mapped_origin) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
610 lax_schema = coerce_instance_wrap(dict_schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
611 strict_schema = core_schema.chain_schema([check_instance, lax_schema]) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
613 schema = core_schema.lax_or_strict_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
614 lax_schema=lax_schema,
615 strict_schema=strict_schema,
616 serialization=core_schema.wrap_serializer_function_ser_schema(
617 lambda v, h: h(v), schema=dict_schema, info_arg=False
618 ),
619 )
621 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
623 def _fraction_schema(self) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
624 """Support for [`fractions.Fraction`][fractions.Fraction]."""
625 from ._validators import fraction_validator 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
627 # TODO: note, this is a fairly common pattern, re lax / strict for attempted type coercion,
628 # can we use a helper function to reduce boilerplate?
629 return core_schema.lax_or_strict_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
630 lax_schema=core_schema.no_info_plain_validator_function(fraction_validator),
631 strict_schema=core_schema.json_or_python_schema(
632 json_schema=core_schema.no_info_plain_validator_function(fraction_validator),
633 python_schema=core_schema.is_instance_schema(Fraction),
634 ),
635 # use str serialization to guarantee round trip behavior
636 serialization=core_schema.to_string_ser_schema(when_used='always'),
637 metadata={'pydantic_js_functions': [lambda _1, _2: {'type': 'string', 'format': 'fraction'}]},
638 )
640 def _arbitrary_type_schema(self, tp: Any) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
641 if not isinstance(tp, type): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
642 warnings.warn( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
643 f'{tp!r} is not a Python type (it may be an instance of an object),'
644 ' Pydantic will allow any object with no validation since we cannot even'
645 ' enforce that the input is an instance of the given type.'
646 ' To get rid of this error wrap the type with `pydantic.SkipValidation`.',
647 ArbitraryTypeWarning,
648 )
649 return core_schema.any_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
650 return core_schema.is_instance_schema(tp) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
652 def _unknown_type_schema(self, obj: Any) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
653 raise PydanticSchemaGenerationError( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
654 f'Unable to generate pydantic-core schema for {obj!r}. '
655 'Set `arbitrary_types_allowed=True` in the model_config to ignore this error'
656 ' or implement `__get_pydantic_core_schema__` on your type to fully support it.'
657 '\n\nIf you got this error by calling handler(<some type>) within'
658 ' `__get_pydantic_core_schema__` then you likely need to call'
659 ' `handler.generate_schema(<some type>)` since we do not call'
660 ' `__get_pydantic_core_schema__` on `<some type>` otherwise to avoid infinite recursion.'
661 )
663 def _apply_discriminator_to_union( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
664 self, schema: CoreSchema, discriminator: str | Discriminator | None
665 ) -> CoreSchema:
666 if discriminator is None: 666 ↛ 667line 666 didn't jump to line 667 because the condition on line 666 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
667 return schema
668 try: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
669 return _discriminated_union.apply_discriminator( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
670 schema,
671 discriminator,
672 self.defs._definitions,
673 )
674 except _discriminated_union.MissingDefinitionForUnionRef: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
675 # defer until defs are resolved
676 _discriminated_union.set_discriminator_in_metadata( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
677 schema,
678 discriminator,
679 )
680 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
682 def clean_schema(self, schema: CoreSchema) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
683 return self.defs.finalize_schema(schema) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
685 def _add_js_function(self, metadata_schema: CoreSchema, js_function: Callable[..., Any]) -> None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
686 metadata = metadata_schema.get('metadata', {}) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
687 pydantic_js_functions = metadata.setdefault('pydantic_js_functions', []) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
688 # because of how we generate core schemas for nested generic models
689 # we can end up adding `BaseModel.__get_pydantic_json_schema__` multiple times
690 # this check may fail to catch duplicates if the function is a `functools.partial`
691 # or something like that, but if it does it'll fail by inserting the duplicate
692 if js_function not in pydantic_js_functions: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
693 pydantic_js_functions.append(js_function) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
694 metadata_schema['metadata'] = metadata 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
696 def generate_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
697 self,
698 obj: Any,
699 ) -> core_schema.CoreSchema:
700 """Generate core schema.
702 Args:
703 obj: The object to generate core schema for.
705 Returns:
706 The generated core schema.
708 Raises:
709 PydanticUndefinedAnnotation:
710 If it is not possible to evaluate forward reference.
711 PydanticSchemaGenerationError:
712 If it is not possible to generate pydantic-core schema.
713 TypeError:
714 - If `alias_generator` returns a disallowed type (must be str, AliasPath or AliasChoices).
715 - If V1 style validator with `each_item=True` applied on a wrong field.
716 PydanticUserError:
717 - If `typing.TypedDict` is used instead of `typing_extensions.TypedDict` on Python < 3.12.
718 - If `__modify_schema__` method is used instead of `__get_pydantic_json_schema__`.
719 """
720 schema = self._generate_schema_from_get_schema_method(obj, obj) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
722 if schema is None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
723 schema = self._generate_schema_inner(obj) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
725 metadata_js_function = _extract_get_pydantic_json_schema(obj) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
726 if metadata_js_function is not None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
727 metadata_schema = resolve_original_schema(schema, self.defs) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
728 if metadata_schema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
729 self._add_js_function(metadata_schema, metadata_js_function) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
731 schema = _add_custom_serialization_from_json_encoders(self._config_wrapper.json_encoders, obj, schema) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
733 return schema 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
735 def _model_schema(self, cls: type[BaseModel]) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
736 """Generate schema for a Pydantic model."""
737 BaseModel_ = import_cached_base_model() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
739 with self.defs.get_schema_or_ref(cls) as (model_ref, maybe_schema): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
740 if maybe_schema is not None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
741 return maybe_schema 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
743 schema = cls.__dict__.get('__pydantic_core_schema__') 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
744 if schema is not None and not isinstance(schema, MockCoreSchema): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
745 if schema['type'] == 'definitions': 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
746 schema = self.defs.unpack_definitions(schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
747 ref = get_ref(schema) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
748 if ref: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
749 return self.defs.create_definition_reference_schema(schema) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
750 else:
751 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
753 config_wrapper = ConfigWrapper(cls.model_config, check=False) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
755 with self._config_wrapper_stack.push(config_wrapper), self._ns_resolver.push(cls): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
756 core_config = self._config_wrapper.core_config(title=cls.__name__) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
758 if cls.__pydantic_fields_complete__ or cls is BaseModel_: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
759 fields = getattr(cls, '__pydantic_fields__', {}) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
760 else:
761 if not hasattr(cls, '__pydantic_fields__'): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
762 # This happens when we have a loop in the schema generation:
763 # class Base[T](BaseModel):
764 # t: T
765 #
766 # class Other(BaseModel):
767 # b: 'Base[Other]'
768 # When we build fields for `Other`, we evaluate the forward annotation.
769 # At this point, `Other` doesn't have the model fields set. We create
770 # `Base[Other]`; model fields are successfully built, and we try to generate
771 # a schema for `t: Other`. As `Other.__pydantic_fields__` aren't set, we abort.
772 raise PydanticUndefinedAnnotation( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
773 name=cls.__name__,
774 message=f'Class {cls.__name__!r} is not defined',
775 )
776 try: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
777 fields = rebuild_model_fields( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
778 cls,
779 config_wrapper=self._config_wrapper,
780 ns_resolver=self._ns_resolver,
781 typevars_map=self._typevars_map or {},
782 )
783 except NameError as e: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
784 raise PydanticUndefinedAnnotation.from_name_error(e) from e 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
786 decorators = cls.__pydantic_decorators__ 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
787 computed_fields = decorators.computed_fields 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
788 check_decorator_fields_exist( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
789 chain(
790 decorators.field_validators.values(),
791 decorators.field_serializers.values(),
792 decorators.validators.values(),
793 ),
794 {*fields.keys(), *computed_fields.keys()},
795 )
797 model_validators = decorators.model_validators.values() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
799 extras_schema = None 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
800 extras_keys_schema = None 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
801 if core_config.get('extra_fields_behavior') == 'allow': 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
802 assert cls.__mro__[0] is cls 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
803 assert cls.__mro__[-1] is object 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
804 for candidate_cls in cls.__mro__[:-1]: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
805 extras_annotation = getattr(candidate_cls, '__annotations__', {}).get( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
806 '__pydantic_extra__', None
807 )
808 if extras_annotation is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
809 if isinstance(extras_annotation, str): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
810 extras_annotation = _typing_extra.eval_type_backport( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
811 _typing_extra._make_forward_ref(
812 extras_annotation, is_argument=False, is_class=True
813 ),
814 *self._types_namespace,
815 )
816 tp = get_origin(extras_annotation) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
817 if tp not in DICT_TYPES: 817 ↛ 818line 817 didn't jump to line 818 because the condition on line 817 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
818 raise PydanticSchemaGenerationError(
819 'The type annotation for `__pydantic_extra__` must be `dict[str, ...]`'
820 )
821 extra_keys_type, extra_items_type = self._get_args_resolving_forward_refs( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
822 extras_annotation,
823 required=True,
824 )
825 if extra_keys_type is not str: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
826 extras_keys_schema = self.generate_schema(extra_keys_type) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
827 if not typing_objects.is_any(extra_items_type): 827 ↛ 829line 827 didn't jump to line 829 because the condition on line 827 was always true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
828 extras_schema = self.generate_schema(extra_items_type) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
829 if extras_keys_schema is not None or extras_schema is not None: 829 ↛ 804line 829 didn't jump to line 804 because the condition on line 829 was always true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
830 break 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
832 generic_origin: type[BaseModel] | None = getattr(cls, '__pydantic_generic_metadata__', {}).get('origin') 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
834 if cls.__pydantic_root_model__: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
835 root_field = self._common_field_schema('root', fields['root'], decorators) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
836 inner_schema = root_field['schema'] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
837 inner_schema = apply_model_validators(inner_schema, model_validators, 'inner') 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
838 model_schema = core_schema.model_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
839 cls,
840 inner_schema,
841 generic_origin=generic_origin,
842 custom_init=getattr(cls, '__pydantic_custom_init__', None),
843 root_model=True,
844 post_init=getattr(cls, '__pydantic_post_init__', None),
845 config=core_config,
846 ref=model_ref,
847 )
848 else:
849 fields_schema: core_schema.CoreSchema = core_schema.model_fields_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
850 {k: self._generate_md_field_schema(k, v, decorators) for k, v in fields.items()},
851 computed_fields=[
852 self._computed_field_schema(d, decorators.field_serializers)
853 for d in computed_fields.values()
854 ],
855 extras_schema=extras_schema,
856 extras_keys_schema=extras_keys_schema,
857 model_name=cls.__name__,
858 )
859 inner_schema = apply_validators(fields_schema, decorators.root_validators.values()) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
860 inner_schema = apply_model_validators(inner_schema, model_validators, 'inner') 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
862 model_schema = core_schema.model_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
863 cls,
864 inner_schema,
865 generic_origin=generic_origin,
866 custom_init=getattr(cls, '__pydantic_custom_init__', None),
867 root_model=False,
868 post_init=getattr(cls, '__pydantic_post_init__', None),
869 config=core_config,
870 ref=model_ref,
871 )
873 schema = self._apply_model_serializers(model_schema, decorators.model_serializers.values()) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
874 schema = apply_model_validators(schema, model_validators, 'outer') 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
875 return self.defs.create_definition_reference_schema(schema) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
877 def _resolve_self_type(self, obj: Any) -> Any: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
878 obj = self.model_type_stack.get() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
879 if obj is None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
880 raise PydanticUserError('`typing.Self` is invalid in this context', code='invalid-self-type') 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
881 return obj 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
883 def _generate_schema_from_get_schema_method(self, obj: Any, source: Any) -> core_schema.CoreSchema | None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
884 BaseModel_ = import_cached_base_model() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
886 get_schema = getattr(obj, '__get_pydantic_core_schema__', None) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
887 is_base_model_get_schema = ( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
888 getattr(get_schema, '__func__', None) is BaseModel_.__get_pydantic_core_schema__.__func__ # pyright: ignore[reportFunctionMemberAccess]
889 )
891 if ( 1abqfklim
892 get_schema is not None
893 # BaseModel.__get_pydantic_core_schema__ is defined for backwards compatibility,
894 # to allow existing code to call `super().__get_pydantic_core_schema__` in Pydantic
895 # model that overrides `__get_pydantic_core_schema__`. However, it raises a deprecation
896 # warning stating that the method will be removed, and during the core schema gen we actually
897 # don't call the method:
898 and not is_base_model_get_schema
899 ):
900 # Some referenceable types might have a `__get_pydantic_core_schema__` method
901 # defined on it by users (e.g. on a dataclass). This generally doesn't play well
902 # as these types are already recognized by the `GenerateSchema` class and isn't ideal
903 # as we might end up calling `get_schema_or_ref` (expensive) on types that are actually
904 # not referenceable:
905 with self.defs.get_schema_or_ref(obj) as (_, maybe_schema): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
906 if maybe_schema is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
907 return maybe_schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
909 if obj is source: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
910 ref_mode = 'unpack' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
911 else:
912 ref_mode = 'to-def' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
913 schema = get_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
914 source, CallbackGetCoreSchemaHandler(self._generate_schema_inner, self, ref_mode=ref_mode)
915 )
916 if schema['type'] == 'definitions': 916 ↛ 917line 916 didn't jump to line 917 because the condition on line 916 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
917 schema = self.defs.unpack_definitions(schema)
919 ref = get_ref(schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
920 if ref: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
921 return self.defs.create_definition_reference_schema(schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
923 # Note: if schema is of type `'definition-ref'`, we might want to copy it as a
924 # safety measure (because these are inlined in place -- i.e. mutated directly)
925 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
927 if get_schema is None and (validators := getattr(obj, '__get_validators__', None)) is not None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
928 from pydantic.v1 import BaseModel as BaseModelV1 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
930 if issubclass(obj, BaseModelV1): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
931 warnings.warn( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
932 f'Mixing V1 models and V2 models (or constructs, like `TypeAdapter`) is not supported. Please upgrade `{obj.__name__}` to V2.',
933 UserWarning,
934 )
935 else:
936 warnings.warn( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
937 '`__get_validators__` is deprecated and will be removed, use `__get_pydantic_core_schema__` instead.',
938 PydanticDeprecatedSince20,
939 )
940 return core_schema.chain_schema([core_schema.with_info_plain_validator_function(v) for v in validators()]) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
942 def _resolve_forward_ref(self, obj: Any) -> Any: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
943 # we assume that types_namespace has the target of forward references in its scope,
944 # but this could fail, for example, if calling Validator on an imported type which contains
945 # forward references to other types only defined in the module from which it was imported
946 # `Validator(SomeImportedTypeAliasWithAForwardReference)`
947 # or the equivalent for BaseModel
948 # class Model(BaseModel):
949 # x: SomeImportedTypeAliasWithAForwardReference
950 try: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
951 obj = _typing_extra.eval_type_backport(obj, *self._types_namespace) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
952 except NameError as e: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
953 raise PydanticUndefinedAnnotation.from_name_error(e) from e 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
955 # if obj is still a ForwardRef, it means we can't evaluate it, raise PydanticUndefinedAnnotation
956 if isinstance(obj, ForwardRef): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
957 raise PydanticUndefinedAnnotation(obj.__forward_arg__, f'Unable to evaluate forward reference {obj}') 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
959 if self._typevars_map: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
960 obj = replace_types(obj, self._typevars_map) 1abceqfklghimdj
962 return obj 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
964 @overload 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
965 def _get_args_resolving_forward_refs(self, obj: Any, required: Literal[True]) -> tuple[Any, ...]: ... 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
967 @overload 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
968 def _get_args_resolving_forward_refs(self, obj: Any) -> tuple[Any, ...] | None: ... 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
970 def _get_args_resolving_forward_refs(self, obj: Any, required: bool = False) -> tuple[Any, ...] | None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
971 args = get_args(obj) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
972 if args: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
973 if isinstance(obj, GenericAlias): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
974 # PEP 585 generic aliases don't convert args to ForwardRefs, unlike `typing.List/Dict` etc.
975 args = (_typing_extra._make_forward_ref(a) if isinstance(a, str) else a for a in args) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
976 args = tuple(self._resolve_forward_ref(a) if isinstance(a, ForwardRef) else a for a in args) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
977 elif required: # pragma: no cover 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
978 raise TypeError(f'Expected {obj} to have generic parameters but it had none')
979 return args 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
981 def _get_first_arg_or_any(self, obj: Any) -> Any: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
982 args = self._get_args_resolving_forward_refs(obj) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
983 if not args: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
984 return Any 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
985 return args[0] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
987 def _get_first_two_args_or_any(self, obj: Any) -> tuple[Any, Any]: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
988 args = self._get_args_resolving_forward_refs(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
989 if not args: 989 ↛ 990line 989 didn't jump to line 990 because the condition on line 989 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
990 return (Any, Any)
991 if len(args) < 2: 991 ↛ 992line 991 didn't jump to line 992 because the condition on line 991 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
992 origin = get_origin(obj)
993 raise TypeError(f'Expected two type arguments for {origin}, got 1')
994 return args[0], args[1] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
996 def _generate_schema_inner(self, obj: Any) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
997 if typing_objects.is_self(obj): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
998 obj = self._resolve_self_type(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1000 if typing_objects.is_annotated(get_origin(obj)): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1001 return self._annotated_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1003 if isinstance(obj, dict): 1003 ↛ 1005line 1003 didn't jump to line 1005 because the condition on line 1003 was never true1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1004 # we assume this is already a valid schema
1005 return obj # type: ignore[return-value]
1007 if isinstance(obj, str): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1008 obj = ForwardRef(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1010 if isinstance(obj, ForwardRef): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1011 return self.generate_schema(self._resolve_forward_ref(obj)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1013 BaseModel = import_cached_base_model() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1015 if lenient_issubclass(obj, BaseModel): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1016 with self.model_type_stack.push(obj): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1017 return self._model_schema(obj) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1019 if isinstance(obj, PydanticRecursiveRef): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1020 return core_schema.definition_reference_schema(schema_ref=obj.type_ref) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1022 return self.match_type(obj) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1024 def match_type(self, obj: Any) -> core_schema.CoreSchema: # noqa: C901 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1025 """Main mapping of types to schemas.
1027 The general structure is a series of if statements starting with the simple cases
1028 (non-generic primitive types) and then handling generics and other more complex cases.
1030 Each case either generates a schema directly, calls into a public user-overridable method
1031 (like `GenerateSchema.tuple_variable_schema`) or calls into a private method that handles some
1032 boilerplate before calling into the user-facing method (e.g. `GenerateSchema._tuple_schema`).
1034 The idea is that we'll evolve this into adding more and more user facing methods over time
1035 as they get requested and we figure out what the right API for them is.
1036 """
1037 if obj is str: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1038 return core_schema.str_schema() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1039 elif obj is bytes: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1040 return core_schema.bytes_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1041 elif obj is int: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1042 return core_schema.int_schema() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1043 elif obj is float: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1044 return core_schema.float_schema() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1045 elif obj is bool: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1046 return core_schema.bool_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1047 elif obj is complex: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1048 return core_schema.complex_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1049 elif typing_objects.is_any(obj) or obj is object: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1050 return core_schema.any_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1051 elif obj is datetime.date: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1052 return core_schema.date_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1053 elif obj is datetime.datetime: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1054 return core_schema.datetime_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1055 elif obj is datetime.time: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1056 return core_schema.time_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1057 elif obj is datetime.timedelta: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1058 return core_schema.timedelta_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1059 elif obj is Decimal: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1060 return core_schema.decimal_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1061 elif obj is UUID: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1062 return core_schema.uuid_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1063 elif obj is Url: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1064 return core_schema.url_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1065 elif obj is Fraction: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1066 return self._fraction_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1067 elif obj is MultiHostUrl: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1068 return core_schema.multi_host_url_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1069 elif obj is None or obj is _typing_extra.NoneType: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1070 return core_schema.none_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1071 elif obj in IP_TYPES: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1072 return self._ip_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1073 elif obj in TUPLE_TYPES: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1074 return self._tuple_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1075 elif obj in LIST_TYPES: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1076 return self._list_schema(Any) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1077 elif obj in SET_TYPES: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1078 return self._set_schema(Any) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1079 elif obj in FROZEN_SET_TYPES: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1080 return self._frozenset_schema(Any) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1081 elif obj in SEQUENCE_TYPES: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1082 return self._sequence_schema(Any) 1acnuAidovB
1083 elif obj in ITERABLE_TYPES: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1084 return self._iterable_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1085 elif obj in DICT_TYPES: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1086 return self._dict_schema(Any, Any) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1087 elif obj in PATH_TYPES: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1088 return self._path_schema(obj, Any) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1089 elif obj in DEQUE_TYPES: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1090 return self._deque_schema(Any) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1091 elif obj in MAPPING_TYPES: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1092 return self._mapping_schema(obj, Any, Any) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1093 elif obj in COUNTER_TYPES: 1093 ↛ 1094line 1093 didn't jump to line 1094 because the condition on line 1093 was never true1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1094 return self._mapping_schema(obj, Any, int)
1095 elif typing_objects.is_typealiastype(obj): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1096 return self._type_alias_type_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1097 elif obj is type: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1098 return self._type_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1099 elif _typing_extra.is_callable(obj): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1100 return core_schema.callable_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1101 elif typing_objects.is_literal(get_origin(obj)): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1102 return self._literal_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1103 elif is_typeddict(obj): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1104 return self._typed_dict_schema(obj, None) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1105 elif _typing_extra.is_namedtuple(obj): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1106 return self._namedtuple_schema(obj, None) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1107 elif typing_objects.is_newtype(obj): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1108 # NewType, can't use isinstance because it fails <3.10
1109 return self.generate_schema(obj.__supertype__) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1110 elif obj in PATTERN_TYPES: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1111 return self._pattern_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1112 elif _typing_extra.is_hashable(obj): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1113 return self._hashable_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1114 elif isinstance(obj, typing.TypeVar): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1115 return self._unsubstituted_typevar_schema(obj) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1116 elif _typing_extra.is_finalvar(obj): 1116 ↛ 1117line 1116 didn't jump to line 1117 because the condition on line 1116 was never true1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1117 if obj is Final:
1118 return core_schema.any_schema()
1119 return self.generate_schema(
1120 self._get_first_arg_or_any(obj),
1121 )
1122 elif isinstance(obj, VALIDATE_CALL_SUPPORTED_TYPES): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1123 return self._call_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1124 elif inspect.isclass(obj) and issubclass(obj, Enum): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1125 return self._enum_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1126 elif obj is ZoneInfo: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1127 return self._zoneinfo_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1129 # dataclasses.is_dataclass coerces dc instances to types, but we only handle
1130 # the case of a dc type here
1131 if dataclasses.is_dataclass(obj): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1132 return self._dataclass_schema(obj, None) # pyright: ignore[reportArgumentType] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1134 origin = get_origin(obj) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1135 if origin is not None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1136 return self._match_generic_type(obj, origin) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1138 if self._arbitrary_types: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1139 return self._arbitrary_type_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1140 return self._unknown_type_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1142 def _match_generic_type(self, obj: Any, origin: Any) -> CoreSchema: # noqa: C901 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1143 # Need to handle generic dataclasses before looking for the schema properties because attribute accesses
1144 # on _GenericAlias delegate to the origin type, so lose the information about the concrete parametrization
1145 # As a result, currently, there is no way to cache the schema for generic dataclasses. This may be possible
1146 # to resolve by modifying the value returned by `Generic.__class_getitem__`, but that is a dangerous game.
1147 if dataclasses.is_dataclass(origin): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1148 return self._dataclass_schema(obj, origin) # pyright: ignore[reportArgumentType] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1149 if _typing_extra.is_namedtuple(origin): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1150 return self._namedtuple_schema(obj, origin) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1152 schema = self._generate_schema_from_get_schema_method(origin, obj) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1153 if schema is not None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1154 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1156 if typing_objects.is_typealiastype(origin): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1157 return self._type_alias_type_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1158 elif is_union_origin(origin): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1159 return self._union_schema(obj) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1160 elif origin in TUPLE_TYPES: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1161 return self._tuple_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1162 elif origin in LIST_TYPES: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1163 return self._list_schema(self._get_first_arg_or_any(obj)) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1164 elif origin in SET_TYPES: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1165 return self._set_schema(self._get_first_arg_or_any(obj)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1166 elif origin in FROZEN_SET_TYPES: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1167 return self._frozenset_schema(self._get_first_arg_or_any(obj)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1168 elif origin in DICT_TYPES: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1169 return self._dict_schema(*self._get_first_two_args_or_any(obj)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1170 elif origin in PATH_TYPES: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1171 return self._path_schema(origin, self._get_first_arg_or_any(obj)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1172 elif origin in DEQUE_TYPES: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1173 return self._deque_schema(self._get_first_arg_or_any(obj)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1174 elif origin in MAPPING_TYPES: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1175 return self._mapping_schema(origin, *self._get_first_two_args_or_any(obj)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1176 elif origin in COUNTER_TYPES: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1177 return self._mapping_schema(origin, self._get_first_arg_or_any(obj), int) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1178 elif is_typeddict(origin): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1179 return self._typed_dict_schema(obj, origin) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1180 elif origin in TYPE_TYPES: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1181 return self._subclass_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1182 elif origin in SEQUENCE_TYPES: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1183 return self._sequence_schema(self._get_first_arg_or_any(obj)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1184 elif origin in ITERABLE_TYPES: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1185 return self._iterable_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1186 elif origin in PATTERN_TYPES: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1187 return self._pattern_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1189 if self._arbitrary_types: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1190 return self._arbitrary_type_schema(origin) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1191 return self._unknown_type_schema(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1193 def _generate_td_field_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1194 self,
1195 name: str,
1196 field_info: FieldInfo,
1197 decorators: DecoratorInfos,
1198 *,
1199 required: bool = True,
1200 ) -> core_schema.TypedDictField:
1201 """Prepare a TypedDictField to represent a model or typeddict field."""
1202 common_field = self._common_field_schema(name, field_info, decorators) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1203 return core_schema.typed_dict_field( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1204 common_field['schema'],
1205 required=False if not field_info.is_required() else required,
1206 serialization_exclude=common_field['serialization_exclude'],
1207 validation_alias=common_field['validation_alias'],
1208 serialization_alias=common_field['serialization_alias'],
1209 metadata=common_field['metadata'],
1210 )
1212 def _generate_md_field_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1213 self,
1214 name: str,
1215 field_info: FieldInfo,
1216 decorators: DecoratorInfos,
1217 ) -> core_schema.ModelField:
1218 """Prepare a ModelField to represent a model field."""
1219 common_field = self._common_field_schema(name, field_info, decorators) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1220 return core_schema.model_field( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1221 common_field['schema'],
1222 serialization_exclude=common_field['serialization_exclude'],
1223 validation_alias=common_field['validation_alias'],
1224 serialization_alias=common_field['serialization_alias'],
1225 frozen=common_field['frozen'],
1226 metadata=common_field['metadata'],
1227 )
1229 def _generate_dc_field_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1230 self,
1231 name: str,
1232 field_info: FieldInfo,
1233 decorators: DecoratorInfos,
1234 ) -> core_schema.DataclassField:
1235 """Prepare a DataclassField to represent the parameter/field, of a dataclass."""
1236 common_field = self._common_field_schema(name, field_info, decorators) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1237 return core_schema.dataclass_field( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1238 name,
1239 common_field['schema'],
1240 init=field_info.init,
1241 init_only=field_info.init_var or None,
1242 kw_only=None if field_info.kw_only else False,
1243 serialization_exclude=common_field['serialization_exclude'],
1244 validation_alias=common_field['validation_alias'],
1245 serialization_alias=common_field['serialization_alias'],
1246 frozen=common_field['frozen'],
1247 metadata=common_field['metadata'],
1248 )
1250 def _common_field_schema( # C901 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1251 self, name: str, field_info: FieldInfo, decorators: DecoratorInfos
1252 ) -> _CommonField:
1253 source_type, annotations = field_info.annotation, field_info.metadata 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1255 def set_discriminator(schema: CoreSchema) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1256 schema = self._apply_discriminator_to_union(schema, field_info.discriminator) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1257 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1259 # Convert `@field_validator` decorators to `Before/After/Plain/WrapValidator` instances:
1260 validators_from_decorators = [] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1261 for decorator in filter_field_decorator_info_by_field(decorators.field_validators.values(), name): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1262 validators_from_decorators.append(_mode_to_validator[decorator.info.mode]._from_decorator(decorator)) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1264 with self.field_name_stack.push(name): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1265 if field_info.discriminator is not None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1266 schema = self._apply_annotations( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1267 source_type, annotations + validators_from_decorators, transform_inner_schema=set_discriminator
1268 )
1269 else:
1270 schema = self._apply_annotations( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1271 source_type,
1272 annotations + validators_from_decorators,
1273 )
1275 # This V1 compatibility shim should eventually be removed
1276 # push down any `each_item=True` validators
1277 # note that this won't work for any Annotated types that get wrapped by a function validator
1278 # but that's okay because that didn't exist in V1
1279 this_field_validators = filter_field_decorator_info_by_field(decorators.validators.values(), name) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1280 if _validators_require_validate_default(this_field_validators): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1281 field_info.validate_default = True 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1282 each_item_validators = [v for v in this_field_validators if v.info.each_item is True] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1283 this_field_validators = [v for v in this_field_validators if v not in each_item_validators] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1284 schema = apply_each_item_validators(schema, each_item_validators) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1286 schema = apply_validators(schema, this_field_validators) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1288 # the default validator needs to go outside of any other validators
1289 # so that it is the topmost validator for the field validator
1290 # which uses it to check if the field has a default value or not
1291 if not field_info.is_required(): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1292 schema = wrap_default(field_info, schema) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1294 schema = self._apply_field_serializers( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1295 schema, filter_field_decorator_info_by_field(decorators.field_serializers.values(), name)
1296 )
1298 pydantic_js_updates, pydantic_js_extra = _extract_json_schema_info_from_field_info(field_info) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1299 core_metadata: dict[str, Any] = {} 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1300 update_core_metadata( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1301 core_metadata, pydantic_js_updates=pydantic_js_updates, pydantic_js_extra=pydantic_js_extra
1302 )
1304 if isinstance(field_info.validation_alias, (AliasChoices, AliasPath)): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1305 validation_alias = field_info.validation_alias.convert_to_aliases() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1306 else:
1307 validation_alias = field_info.validation_alias 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1309 return _common_field( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1310 schema,
1311 serialization_exclude=True if field_info.exclude else None,
1312 validation_alias=validation_alias,
1313 serialization_alias=field_info.serialization_alias,
1314 frozen=field_info.frozen,
1315 metadata=core_metadata,
1316 )
1318 def _union_schema(self, union_type: Any) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1319 """Generate schema for a Union."""
1320 args = self._get_args_resolving_forward_refs(union_type, required=True) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1321 choices: list[CoreSchema] = [] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1322 nullable = False 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1323 for arg in args: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1324 if arg is None or arg is _typing_extra.NoneType: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1325 nullable = True 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1326 else:
1327 choices.append(self.generate_schema(arg)) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1329 if len(choices) == 1: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1330 s = choices[0] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1331 else:
1332 choices_with_tags: list[CoreSchema | tuple[CoreSchema, str]] = [] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1333 for choice in choices: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1334 tag = cast(CoreMetadata, choice.get('metadata', {})).get('pydantic_internal_union_tag_key') 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1335 if tag is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1336 choices_with_tags.append((choice, tag)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1337 else:
1338 choices_with_tags.append(choice) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1339 s = core_schema.union_schema(choices_with_tags) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1341 if nullable: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1342 s = core_schema.nullable_schema(s) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1343 return s 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1345 def _type_alias_type_schema(self, obj: TypeAliasType) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1346 with self.defs.get_schema_or_ref(obj) as (ref, maybe_schema): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1347 if maybe_schema is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1348 return maybe_schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1350 origin: TypeAliasType = get_origin(obj) or obj 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1351 typevars_map = get_standard_typevars_map(obj) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1353 with self._ns_resolver.push(origin): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1354 try: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1355 annotation = _typing_extra.eval_type(origin.__value__, *self._types_namespace) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1356 except NameError as e:
1357 raise PydanticUndefinedAnnotation.from_name_error(e) from e
1358 annotation = replace_types(annotation, typevars_map) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1359 schema = self.generate_schema(annotation) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1360 assert schema['type'] != 'definitions' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1361 schema['ref'] = ref # type: ignore 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1362 return self.defs.create_definition_reference_schema(schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1364 def _literal_schema(self, literal_type: Any) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1365 """Generate schema for a Literal."""
1366 expected = list(get_literal_values(literal_type, type_check=False, unpack_type_aliases='eager')) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1367 assert expected, f'literal "expected" cannot be empty, obj={literal_type}' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1368 schema = core_schema.literal_schema(expected) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1370 if self._config_wrapper.use_enum_values and any(isinstance(v, Enum) for v in expected): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1371 schema = core_schema.no_info_after_validator_function( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1372 lambda v: v.value if isinstance(v, Enum) else v, schema
1373 )
1375 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1377 def _typed_dict_schema(self, typed_dict_cls: Any, origin: Any) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1378 """Generate a core schema for a `TypedDict` class.
1380 To be able to build a `DecoratorInfos` instance for the `TypedDict` class (which will include
1381 validators, serializers, etc.), we need to have access to the original bases of the class
1382 (see https://docs.python.org/3/library/types.html#types.get_original_bases).
1383 However, the `__orig_bases__` attribute was only added in 3.12 (https://github.com/python/cpython/pull/103698).
1385 For this reason, we require Python 3.12 (or using the `typing_extensions` backport).
1386 """
1387 FieldInfo = import_cached_field_info() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1389 with ( 1abcenpqfklghrsimdjot
1390 self.model_type_stack.push(typed_dict_cls),
1391 self.defs.get_schema_or_ref(typed_dict_cls) as (
1392 typed_dict_ref,
1393 maybe_schema,
1394 ),
1395 ):
1396 if maybe_schema is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1397 return maybe_schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1399 typevars_map = get_standard_typevars_map(typed_dict_cls) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1400 if origin is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1401 typed_dict_cls = origin 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1403 if not _SUPPORTS_TYPEDDICT and type(typed_dict_cls).__module__ == 'typing': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1404 raise PydanticUserError( 1abcenpqfklghrsimdjot
1405 'Please use `typing_extensions.TypedDict` instead of `typing.TypedDict` on Python < 3.12.',
1406 code='typed-dict-version',
1407 )
1409 try: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1410 # if a typed dictionary class doesn't have config, we use the parent's config, hence a default of `None`
1411 # see https://github.com/pydantic/pydantic/issues/10917
1412 config: ConfigDict | None = get_attribute_from_bases(typed_dict_cls, '__pydantic_config__') 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1413 except AttributeError: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1414 config = None 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1416 with self._config_wrapper_stack.push(config): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1417 core_config = self._config_wrapper.core_config(title=typed_dict_cls.__name__) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1419 required_keys: frozenset[str] = typed_dict_cls.__required_keys__ 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1421 fields: dict[str, core_schema.TypedDictField] = {} 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1423 decorators = DecoratorInfos.build(typed_dict_cls) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1424 decorators.update_from_config(self._config_wrapper) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1426 if self._config_wrapper.use_attribute_docstrings: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1427 field_docstrings = extract_docstrings_from_cls(typed_dict_cls, use_inspect=True) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1428 else:
1429 field_docstrings = None 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1431 try: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1432 annotations = _typing_extra.get_cls_type_hints(typed_dict_cls, ns_resolver=self._ns_resolver) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1433 except NameError as e:
1434 raise PydanticUndefinedAnnotation.from_name_error(e) from e
1436 readonly_fields: list[str] = [] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1438 for field_name, annotation in annotations.items(): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1439 field_info = FieldInfo.from_annotation(annotation, _source=AnnotationSource.TYPED_DICT) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1440 field_info.annotation = replace_types(field_info.annotation, typevars_map) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1442 required = ( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1443 field_name in required_keys or 'required' in field_info._qualifiers
1444 ) and 'not_required' not in field_info._qualifiers
1445 if 'read_only' in field_info._qualifiers: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1446 readonly_fields.append(field_name) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1448 if ( 1abqfklim
1449 field_docstrings is not None
1450 and field_info.description is None
1451 and field_name in field_docstrings
1452 ):
1453 field_info.description = field_docstrings[field_name] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1454 update_field_from_config(self._config_wrapper, field_name, field_info) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1456 fields[field_name] = self._generate_td_field_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1457 field_name, field_info, decorators, required=required
1458 )
1460 if readonly_fields: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1461 fields_repr = ', '.join(repr(f) for f in readonly_fields) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1462 plural = len(readonly_fields) >= 2 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1463 warnings.warn( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1464 f'Item{"s" if plural else ""} {fields_repr} on TypedDict class {typed_dict_cls.__name__!r} '
1465 f'{"are" if plural else "is"} using the `ReadOnly` qualifier. Pydantic will not protect items '
1466 'from any mutation on dictionary instances.',
1467 UserWarning,
1468 )
1470 td_schema = core_schema.typed_dict_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1471 fields,
1472 cls=typed_dict_cls,
1473 computed_fields=[
1474 self._computed_field_schema(d, decorators.field_serializers)
1475 for d in decorators.computed_fields.values()
1476 ],
1477 ref=typed_dict_ref,
1478 config=core_config,
1479 )
1481 schema = self._apply_model_serializers(td_schema, decorators.model_serializers.values()) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1482 schema = apply_model_validators(schema, decorators.model_validators.values(), 'all') 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1483 return self.defs.create_definition_reference_schema(schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1485 def _namedtuple_schema(self, namedtuple_cls: Any, origin: Any) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1486 """Generate schema for a NamedTuple."""
1487 with ( 1abcenpqfklghrsimdjot
1488 self.model_type_stack.push(namedtuple_cls),
1489 self.defs.get_schema_or_ref(namedtuple_cls) as (
1490 namedtuple_ref,
1491 maybe_schema,
1492 ),
1493 ):
1494 if maybe_schema is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1495 return maybe_schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1496 typevars_map = get_standard_typevars_map(namedtuple_cls) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1497 if origin is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1498 namedtuple_cls = origin 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1500 try: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1501 annotations = _typing_extra.get_cls_type_hints(namedtuple_cls, ns_resolver=self._ns_resolver) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1502 except NameError as e:
1503 raise PydanticUndefinedAnnotation.from_name_error(e) from e
1504 if not annotations: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1505 # annotations is empty, happens if namedtuple_cls defined via collections.namedtuple(...)
1506 annotations: dict[str, Any] = {k: Any for k in namedtuple_cls._fields} 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1508 if typevars_map: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1509 annotations = { 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1510 field_name: replace_types(annotation, typevars_map)
1511 for field_name, annotation in annotations.items()
1512 }
1514 arguments_schema = core_schema.arguments_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1515 [
1516 self._generate_parameter_schema(
1517 field_name,
1518 annotation,
1519 source=AnnotationSource.NAMED_TUPLE,
1520 default=namedtuple_cls._field_defaults.get(field_name, Parameter.empty),
1521 )
1522 for field_name, annotation in annotations.items()
1523 ],
1524 metadata={'pydantic_js_prefer_positional_arguments': True},
1525 )
1526 schema = core_schema.call_schema(arguments_schema, namedtuple_cls, ref=namedtuple_ref) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1527 return self.defs.create_definition_reference_schema(schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1529 def _generate_parameter_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1530 self,
1531 name: str,
1532 annotation: type[Any],
1533 source: AnnotationSource,
1534 default: Any = Parameter.empty,
1535 mode: Literal['positional_only', 'positional_or_keyword', 'keyword_only'] | None = None,
1536 ) -> core_schema.ArgumentsParameter:
1537 """Generate the definition of a field in a namedtuple or a parameter in a function signature.
1539 This definition is meant to be used for the `'arguments'` core schema, which will be replaced
1540 in V3 by the `'arguments-v3`'.
1541 """
1542 FieldInfo = import_cached_field_info() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1544 if default is Parameter.empty: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1545 field = FieldInfo.from_annotation(annotation, _source=source) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1546 else:
1547 field = FieldInfo.from_annotated_attribute(annotation, default, _source=source) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1549 assert field.annotation is not None, 'field.annotation should not be None when generating a schema' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1550 update_field_from_config(self._config_wrapper, name, field) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1552 with self.field_name_stack.push(name): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1553 schema = self._apply_annotations( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1554 field.annotation,
1555 [field],
1556 # Because we pass `field` as metadata above (required for attributes relevant for
1557 # JSON Scheme generation), we need to ignore the potential warnings about `FieldInfo`
1558 # attributes that will not be used:
1559 check_unsupported_field_info_attributes=False,
1560 )
1562 if not field.is_required(): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1563 schema = wrap_default(field, schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1565 parameter_schema = core_schema.arguments_parameter(name, schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1566 if mode is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1567 parameter_schema['mode'] = mode 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1568 if field.alias is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1569 parameter_schema['alias'] = field.alias 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1571 return parameter_schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1573 def _generate_parameter_v3_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1574 self,
1575 name: str,
1576 annotation: Any,
1577 source: AnnotationSource,
1578 mode: Literal[
1579 'positional_only',
1580 'positional_or_keyword',
1581 'keyword_only',
1582 'var_args',
1583 'var_kwargs_uniform',
1584 'var_kwargs_unpacked_typed_dict',
1585 ],
1586 default: Any = Parameter.empty,
1587 ) -> core_schema.ArgumentsV3Parameter:
1588 """Generate the definition of a parameter in a function signature.
1590 This definition is meant to be used for the `'arguments-v3'` core schema, which will replace
1591 the `'arguments`' schema in V3.
1592 """
1593 FieldInfo = import_cached_field_info() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1595 if default is Parameter.empty: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1596 field = FieldInfo.from_annotation(annotation, _source=source) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1597 else:
1598 field = FieldInfo.from_annotated_attribute(annotation, default, _source=source) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1599 update_field_from_config(self._config_wrapper, name, field) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1601 with self.field_name_stack.push(name): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1602 schema = self._apply_annotations( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1603 field.annotation,
1604 [field],
1605 # Because we pass `field` as metadata above (required for attributes relevant for
1606 # JSON Scheme generation), we need to ignore the potential warnings about `FieldInfo`
1607 # attributes that will not be used:
1608 check_unsupported_field_info_attributes=False,
1609 )
1611 if not field.is_required(): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1612 schema = wrap_default(field, schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1614 parameter_schema = core_schema.arguments_v3_parameter( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1615 name=name,
1616 schema=schema,
1617 mode=mode,
1618 )
1619 if field.alias is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1620 parameter_schema['alias'] = field.alias 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1622 return parameter_schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1624 def _tuple_schema(self, tuple_type: Any) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1625 """Generate schema for a Tuple, e.g. `tuple[int, str]` or `tuple[int, ...]`."""
1626 # TODO: do we really need to resolve type vars here?
1627 typevars_map = get_standard_typevars_map(tuple_type) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1628 params = self._get_args_resolving_forward_refs(tuple_type) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1630 if typevars_map and params: 1630 ↛ 1631line 1630 didn't jump to line 1631 because the condition on line 1630 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1631 params = tuple(replace_types(param, typevars_map) for param in params)
1633 # NOTE: subtle difference: `tuple[()]` gives `params=()`, whereas `typing.Tuple[()]` gives `params=((),)`
1634 # This is only true for <3.11, on Python 3.11+ `typing.Tuple[()]` gives `params=()`
1635 if not params: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1636 if tuple_type in TUPLE_TYPES: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1637 return core_schema.tuple_schema([core_schema.any_schema()], variadic_item_index=0) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1638 else:
1639 # special case for `tuple[()]` which means `tuple[]` - an empty tuple
1640 return core_schema.tuple_schema([]) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1641 elif params[-1] is Ellipsis: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1642 if len(params) == 2: 1642 ↛ 1646line 1642 didn't jump to line 1646 because the condition on line 1642 was always true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1643 return core_schema.tuple_schema([self.generate_schema(params[0])], variadic_item_index=0) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1644 else:
1645 # TODO: something like https://github.com/pydantic/pydantic/issues/5952
1646 raise ValueError('Variable tuples can only have one type')
1647 elif len(params) == 1 and params[0] == (): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1648 # special case for `tuple[()]` which means `tuple[]` - an empty tuple
1649 # NOTE: This conditional can be removed when we drop support for Python 3.10.
1650 return core_schema.tuple_schema([]) 1abceqfklghimdj
1651 else:
1652 return core_schema.tuple_schema([self.generate_schema(param) for param in params]) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1654 def _type_schema(self) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1655 return core_schema.custom_error_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1656 core_schema.is_instance_schema(type),
1657 custom_error_type='is_type',
1658 custom_error_message='Input should be a type',
1659 )
1661 def _zoneinfo_schema(self) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1662 """Generate schema for a zone_info.ZoneInfo object"""
1663 from ._validators import validate_str_is_valid_iana_tz 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1665 metadata = {'pydantic_js_functions': [lambda _1, _2: {'type': 'string', 'format': 'zoneinfo'}]} 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1666 return core_schema.no_info_plain_validator_function( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1667 validate_str_is_valid_iana_tz,
1668 serialization=core_schema.to_string_ser_schema(),
1669 metadata=metadata,
1670 )
1672 def _union_is_subclass_schema(self, union_type: Any) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1673 """Generate schema for `type[Union[X, ...]]`."""
1674 args = self._get_args_resolving_forward_refs(union_type, required=True) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1675 return core_schema.union_schema([self.generate_schema(type[args]) for args in args]) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1677 def _subclass_schema(self, type_: Any) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1678 """Generate schema for a type, e.g. `type[int]`."""
1679 type_param = self._get_first_arg_or_any(type_) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1681 # Assume `type[Annotated[<typ>, ...]]` is equivalent to `type[<typ>]`:
1682 type_param = _typing_extra.annotated_type(type_param) or type_param 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1684 if typing_objects.is_any(type_param): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1685 return self._type_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1686 elif typing_objects.is_typealiastype(type_param): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1687 return self.generate_schema(type[type_param.__value__]) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1688 elif typing_objects.is_typevar(type_param): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1689 if type_param.__bound__: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1690 if is_union_origin(get_origin(type_param.__bound__)): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1691 return self._union_is_subclass_schema(type_param.__bound__) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1692 return core_schema.is_subclass_schema(type_param.__bound__) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1693 elif type_param.__constraints__: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1694 return core_schema.union_schema([self.generate_schema(type[c]) for c in type_param.__constraints__]) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1695 else:
1696 return self._type_schema() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1697 elif is_union_origin(get_origin(type_param)): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1698 return self._union_is_subclass_schema(type_param) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1699 else:
1700 if typing_objects.is_self(type_param): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1701 type_param = self._resolve_self_type(type_param) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1702 if _typing_extra.is_generic_alias(type_param): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1703 raise PydanticUserError( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1704 'Subscripting `type[]` with an already parametrized type is not supported. '
1705 f'Instead of using type[{type_param!r}], use type[{_repr.display_as_type(get_origin(type_param))}].',
1706 code=None,
1707 )
1708 if not inspect.isclass(type_param): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1709 # when using type[None], this doesn't type convert to type[NoneType], and None isn't a class
1710 # so we handle it manually here
1711 if type_param is None: 1711 ↛ 1713line 1711 didn't jump to line 1713 because the condition on line 1711 was always true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1712 return core_schema.is_subclass_schema(_typing_extra.NoneType) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1713 raise TypeError(f'Expected a class, got {type_param!r}')
1714 return core_schema.is_subclass_schema(type_param) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1716 def _sequence_schema(self, items_type: Any) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1717 """Generate schema for a Sequence, e.g. `Sequence[int]`."""
1718 from ._serializers import serialize_sequence_via_list 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1720 item_type_schema = self.generate_schema(items_type) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1721 list_schema = core_schema.list_schema(item_type_schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1723 json_schema = smart_deepcopy(list_schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1724 python_schema = core_schema.is_instance_schema(typing.Sequence, cls_repr='Sequence') 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1725 if not typing_objects.is_any(items_type): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1726 from ._validators import sequence_validator 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1728 python_schema = core_schema.chain_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1729 [python_schema, core_schema.no_info_wrap_validator_function(sequence_validator, list_schema)],
1730 )
1732 serialization = core_schema.wrap_serializer_function_ser_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1733 serialize_sequence_via_list, schema=item_type_schema, info_arg=True
1734 )
1735 return core_schema.json_or_python_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1736 json_schema=json_schema, python_schema=python_schema, serialization=serialization
1737 )
1739 def _iterable_schema(self, type_: Any) -> core_schema.GeneratorSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1740 """Generate a schema for an `Iterable`."""
1741 item_type = self._get_first_arg_or_any(type_) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1743 return core_schema.generator_schema(self.generate_schema(item_type)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1745 def _pattern_schema(self, pattern_type: Any) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1746 from . import _validators 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1748 metadata = {'pydantic_js_functions': [lambda _1, _2: {'type': 'string', 'format': 'regex'}]} 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1749 ser = core_schema.plain_serializer_function_ser_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1750 attrgetter('pattern'), when_used='json', return_schema=core_schema.str_schema()
1751 )
1752 if pattern_type is typing.Pattern or pattern_type is re.Pattern: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1753 # bare type
1754 return core_schema.no_info_plain_validator_function( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1755 _validators.pattern_either_validator, serialization=ser, metadata=metadata
1756 )
1758 param = self._get_args_resolving_forward_refs( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1759 pattern_type,
1760 required=True,
1761 )[0]
1762 if param is str: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1763 return core_schema.no_info_plain_validator_function( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1764 _validators.pattern_str_validator, serialization=ser, metadata=metadata
1765 )
1766 elif param is bytes: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1767 return core_schema.no_info_plain_validator_function( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1768 _validators.pattern_bytes_validator, serialization=ser, metadata=metadata
1769 )
1770 else:
1771 raise PydanticSchemaGenerationError(f'Unable to generate pydantic-core schema for {pattern_type!r}.') 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1773 def _hashable_schema(self) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1774 return core_schema.custom_error_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1775 schema=core_schema.json_or_python_schema(
1776 json_schema=core_schema.chain_schema(
1777 [core_schema.any_schema(), core_schema.is_instance_schema(collections.abc.Hashable)]
1778 ),
1779 python_schema=core_schema.is_instance_schema(collections.abc.Hashable),
1780 ),
1781 custom_error_type='is_hashable',
1782 custom_error_message='Input should be hashable',
1783 )
1785 def _dataclass_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1786 self, dataclass: type[StandardDataclass], origin: type[StandardDataclass] | None
1787 ) -> core_schema.CoreSchema:
1788 """Generate schema for a dataclass."""
1789 with ( 1abcenpqfklghrsimdjot
1790 self.model_type_stack.push(dataclass),
1791 self.defs.get_schema_or_ref(dataclass) as (
1792 dataclass_ref,
1793 maybe_schema,
1794 ),
1795 ):
1796 if maybe_schema is not None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1797 return maybe_schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1799 schema = dataclass.__dict__.get('__pydantic_core_schema__') 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1800 if schema is not None and not isinstance(schema, MockCoreSchema): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1801 if schema['type'] == 'definitions': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1802 schema = self.defs.unpack_definitions(schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1803 ref = get_ref(schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1804 if ref: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1805 return self.defs.create_definition_reference_schema(schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1806 else:
1807 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1809 typevars_map = get_standard_typevars_map(dataclass) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1810 if origin is not None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1811 dataclass = origin 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1813 # if (plain) dataclass doesn't have config, we use the parent's config, hence a default of `None`
1814 # (Pydantic dataclasses have an empty dict config by default).
1815 # see https://github.com/pydantic/pydantic/issues/10917
1816 config = getattr(dataclass, '__pydantic_config__', None) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1818 from ..dataclasses import is_pydantic_dataclass 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1820 with self._ns_resolver.push(dataclass), self._config_wrapper_stack.push(config): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1821 if is_pydantic_dataclass(dataclass): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1822 if dataclass.__pydantic_fields_complete__(): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1823 # Copy the field info instances to avoid mutating the `FieldInfo` instances
1824 # of the generic dataclass generic origin (e.g. `apply_typevars_map` below).
1825 # Note that we don't apply `deepcopy` on `__pydantic_fields__` because we
1826 # don't want to copy the `FieldInfo` attributes:
1827 fields = { 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1828 f_name: copy(field_info) for f_name, field_info in dataclass.__pydantic_fields__.items()
1829 }
1830 if typevars_map: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1831 for field in fields.values(): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1832 field.apply_typevars_map(typevars_map, *self._types_namespace) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1833 else:
1834 try: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1835 fields = rebuild_dataclass_fields( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1836 dataclass,
1837 config_wrapper=self._config_wrapper,
1838 ns_resolver=self._ns_resolver,
1839 typevars_map=typevars_map or {},
1840 )
1841 except NameError as e: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1842 raise PydanticUndefinedAnnotation.from_name_error(e) from e 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1843 else:
1844 fields = collect_dataclass_fields( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1845 dataclass,
1846 typevars_map=typevars_map,
1847 config_wrapper=self._config_wrapper,
1848 )
1850 if self._config_wrapper.extra == 'allow': 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1851 # disallow combination of init=False on a dataclass field and extra='allow' on a dataclass
1852 for field_name, field in fields.items(): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1853 if field.init is False: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1854 raise PydanticUserError( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1855 f'Field {field_name} has `init=False` and dataclass has config setting `extra="allow"`. '
1856 f'This combination is not allowed.',
1857 code='dataclass-init-false-extra-allow',
1858 )
1860 decorators = dataclass.__dict__.get('__pydantic_decorators__') 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1861 if decorators is None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1862 decorators = DecoratorInfos.build(dataclass) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1863 decorators.update_from_config(self._config_wrapper) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1864 # Move kw_only=False args to the start of the list, as this is how vanilla dataclasses work.
1865 # Note that when kw_only is missing or None, it is treated as equivalent to kw_only=True
1866 args = sorted( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1867 (self._generate_dc_field_schema(k, v, decorators) for k, v in fields.items()),
1868 key=lambda a: a.get('kw_only') is not False,
1869 )
1870 has_post_init = hasattr(dataclass, '__post_init__') 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1871 has_slots = hasattr(dataclass, '__slots__') 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1873 args_schema = core_schema.dataclass_args_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1874 dataclass.__name__,
1875 args,
1876 computed_fields=[
1877 self._computed_field_schema(d, decorators.field_serializers)
1878 for d in decorators.computed_fields.values()
1879 ],
1880 collect_init_only=has_post_init,
1881 )
1883 inner_schema = apply_validators(args_schema, decorators.root_validators.values()) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1885 model_validators = decorators.model_validators.values() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1886 inner_schema = apply_model_validators(inner_schema, model_validators, 'inner') 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1888 core_config = self._config_wrapper.core_config(title=dataclass.__name__) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1890 dc_schema = core_schema.dataclass_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1891 dataclass,
1892 inner_schema,
1893 generic_origin=origin,
1894 post_init=has_post_init,
1895 ref=dataclass_ref,
1896 fields=[field.name for field in dataclasses.fields(dataclass)],
1897 slots=has_slots,
1898 config=core_config,
1899 # we don't use a custom __setattr__ for dataclasses, so we must
1900 # pass along the frozen config setting to the pydantic-core schema
1901 frozen=self._config_wrapper_stack.tail.frozen,
1902 )
1903 schema = self._apply_model_serializers(dc_schema, decorators.model_serializers.values()) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1904 schema = apply_model_validators(schema, model_validators, 'outer') 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1905 return self.defs.create_definition_reference_schema(schema) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1907 def _call_schema(self, function: ValidateCallSupportedTypes) -> core_schema.CallSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
1908 """Generate schema for a Callable.
1910 TODO support functional validators once we support them in Config
1911 """
1912 arguments_schema = self._arguments_schema(function) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1914 return_schema: core_schema.CoreSchema | None = None 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1915 config_wrapper = self._config_wrapper 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1916 if config_wrapper.validate_return: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1917 sig = signature(function) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1918 return_hint = sig.return_annotation 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1919 if return_hint is not sig.empty: 1919 ↛ 1926line 1919 didn't jump to line 1926 because the condition on line 1919 was always true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1920 globalns, localns = self._types_namespace 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1921 type_hints = _typing_extra.get_function_type_hints( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1922 function, globalns=globalns, localns=localns, include_keys={'return'}
1923 )
1924 return_schema = self.generate_schema(type_hints['return']) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1926 return core_schema.call_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1927 arguments_schema,
1928 function,
1929 return_schema=return_schema,
1930 )
1932 def _arguments_schema( 1abcenpuwACDklghrsxyEFGJimdjotvzBHI
1933 self, function: ValidateCallSupportedTypes, parameters_callback: ParametersCallback | None = None
1934 ) -> core_schema.ArgumentsSchema:
1935 """Generate schema for a Signature."""
1936 mode_lookup: dict[_ParameterKind, Literal['positional_only', 'positional_or_keyword', 'keyword_only']] = { 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1937 Parameter.POSITIONAL_ONLY: 'positional_only',
1938 Parameter.POSITIONAL_OR_KEYWORD: 'positional_or_keyword',
1939 Parameter.KEYWORD_ONLY: 'keyword_only',
1940 }
1942 sig = signature(function) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1943 globalns, localns = self._types_namespace 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1944 type_hints = _typing_extra.get_function_type_hints(function, globalns=globalns, localns=localns) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1946 arguments_list: list[core_schema.ArgumentsParameter] = [] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1947 var_args_schema: core_schema.CoreSchema | None = None 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1948 var_kwargs_schema: core_schema.CoreSchema | None = None 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1949 var_kwargs_mode: core_schema.VarKwargsMode | None = None 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1951 for i, (name, p) in enumerate(sig.parameters.items()): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1952 if p.annotation is sig.empty: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1953 annotation = typing.cast(Any, Any) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1954 else:
1955 annotation = type_hints[name] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1957 if parameters_callback is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1958 result = parameters_callback(i, name, annotation) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1959 if result == 'skip': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1960 continue 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1962 parameter_mode = mode_lookup.get(p.kind) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1963 if parameter_mode is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1964 arg_schema = self._generate_parameter_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1965 name, annotation, AnnotationSource.FUNCTION, p.default, parameter_mode
1966 )
1967 arguments_list.append(arg_schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1968 elif p.kind == Parameter.VAR_POSITIONAL: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1969 var_args_schema = self.generate_schema(annotation) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1970 else:
1971 assert p.kind == Parameter.VAR_KEYWORD, p.kind 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1973 unpack_type = _typing_extra.unpack_type(annotation) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1974 if unpack_type is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1975 origin = get_origin(unpack_type) or unpack_type 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1976 if not is_typeddict(origin): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1977 raise PydanticUserError( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1978 f'Expected a `TypedDict` class inside `Unpack[...]`, got {unpack_type!r}',
1979 code='unpack-typed-dict',
1980 )
1981 non_pos_only_param_names = { 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1982 name for name, p in sig.parameters.items() if p.kind != Parameter.POSITIONAL_ONLY
1983 }
1984 overlapping_params = non_pos_only_param_names.intersection(origin.__annotations__) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1985 if overlapping_params: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1986 raise PydanticUserError( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1987 f'Typed dictionary {origin.__name__!r} overlaps with parameter'
1988 f'{"s" if len(overlapping_params) >= 2 else ""} '
1989 f'{", ".join(repr(p) for p in sorted(overlapping_params))}',
1990 code='overlapping-unpack-typed-dict',
1991 )
1993 var_kwargs_mode = 'unpacked-typed-dict' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1994 var_kwargs_schema = self._typed_dict_schema(unpack_type, get_origin(unpack_type)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1995 else:
1996 var_kwargs_mode = 'uniform' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1997 var_kwargs_schema = self.generate_schema(annotation) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
1999 return core_schema.arguments_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2000 arguments_list,
2001 var_args_schema=var_args_schema,
2002 var_kwargs_mode=var_kwargs_mode,
2003 var_kwargs_schema=var_kwargs_schema,
2004 validate_by_name=self._config_wrapper.validate_by_name,
2005 )
2007 def _arguments_v3_schema( 1abcenpuwACDklghrsxyEFGJimdjotvzBHI
2008 self, function: ValidateCallSupportedTypes, parameters_callback: ParametersCallback | None = None
2009 ) -> core_schema.ArgumentsV3Schema:
2010 mode_lookup: dict[ 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2011 _ParameterKind, Literal['positional_only', 'positional_or_keyword', 'var_args', 'keyword_only']
2012 ] = {
2013 Parameter.POSITIONAL_ONLY: 'positional_only',
2014 Parameter.POSITIONAL_OR_KEYWORD: 'positional_or_keyword',
2015 Parameter.VAR_POSITIONAL: 'var_args',
2016 Parameter.KEYWORD_ONLY: 'keyword_only',
2017 }
2019 sig = signature(function) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2020 globalns, localns = self._types_namespace 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2021 type_hints = _typing_extra.get_function_type_hints(function, globalns=globalns, localns=localns) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2023 parameters_list: list[core_schema.ArgumentsV3Parameter] = [] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2025 for i, (name, p) in enumerate(sig.parameters.items()): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2026 if parameters_callback is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2027 result = parameters_callback(i, name, p.annotation) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2028 if result == 'skip': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2029 continue 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2031 if p.annotation is Parameter.empty: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2032 annotation = typing.cast(Any, Any) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2033 else:
2034 annotation = type_hints[name] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2036 parameter_mode = mode_lookup.get(p.kind) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2037 if parameter_mode is None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2038 assert p.kind == Parameter.VAR_KEYWORD, p.kind 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2040 unpack_type = _typing_extra.unpack_type(annotation) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2041 if unpack_type is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2042 origin = get_origin(unpack_type) or unpack_type 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2043 if not is_typeddict(origin): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2044 raise PydanticUserError( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2045 f'Expected a `TypedDict` class inside `Unpack[...]`, got {unpack_type!r}',
2046 code='unpack-typed-dict',
2047 )
2048 non_pos_only_param_names = { 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2049 name for name, p in sig.parameters.items() if p.kind != Parameter.POSITIONAL_ONLY
2050 }
2051 overlapping_params = non_pos_only_param_names.intersection(origin.__annotations__) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2052 if overlapping_params: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2053 raise PydanticUserError( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2054 f'Typed dictionary {origin.__name__!r} overlaps with parameter'
2055 f'{"s" if len(overlapping_params) >= 2 else ""} '
2056 f'{", ".join(repr(p) for p in sorted(overlapping_params))}',
2057 code='overlapping-unpack-typed-dict',
2058 )
2059 parameter_mode = 'var_kwargs_unpacked_typed_dict' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2060 annotation = unpack_type 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2061 else:
2062 parameter_mode = 'var_kwargs_uniform' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2064 parameters_list.append( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2065 self._generate_parameter_v3_schema(
2066 name, annotation, AnnotationSource.FUNCTION, parameter_mode, default=p.default
2067 )
2068 )
2070 return core_schema.arguments_v3_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2071 parameters_list,
2072 validate_by_name=self._config_wrapper.validate_by_name,
2073 )
2075 def _unsubstituted_typevar_schema(self, typevar: typing.TypeVar) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2076 try: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2077 has_default = typevar.has_default() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2078 except AttributeError: 1abcenpuwqfklghrsxyimdjotvz
2079 # Happens if using `typing.TypeVar` (and not `typing_extensions`) on Python < 3.13
2080 pass 1abcenpuwqfklghrsxyimdjotvz
2081 else:
2082 if has_default: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2083 return self.generate_schema(typevar.__default__) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2085 if constraints := typevar.__constraints__: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2086 return self._union_schema(typing.Union[constraints]) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2088 if bound := typevar.__bound__: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2089 schema = self.generate_schema(bound) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2090 schema['serialization'] = core_schema.wrap_serializer_function_ser_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2091 lambda x, h: h(x),
2092 schema=core_schema.any_schema(),
2093 )
2094 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2096 return core_schema.any_schema() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2098 def _computed_field_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2099 self,
2100 d: Decorator[ComputedFieldInfo],
2101 field_serializers: dict[str, Decorator[FieldSerializerDecoratorInfo]],
2102 ) -> core_schema.ComputedField:
2103 if d.info.return_type is not PydanticUndefined: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2104 return_type = d.info.return_type 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2105 else:
2106 try: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2107 # Do not pass in globals as the function could be defined in a different module.
2108 # Instead, let `get_callable_return_type` infer the globals to use, but still pass
2109 # in locals that may contain a parent/rebuild namespace:
2110 return_type = _decorators.get_callable_return_type(d.func, localns=self._types_namespace.locals) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2111 except NameError as e: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2112 raise PydanticUndefinedAnnotation.from_name_error(e) from e 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2113 if return_type is PydanticUndefined: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2114 raise PydanticUserError( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2115 'Computed field is missing return type annotation or specifying `return_type`'
2116 ' to the `@computed_field` decorator (e.g. `@computed_field(return_type=int | str)`)',
2117 code='model-field-missing-annotation',
2118 )
2120 return_type = replace_types(return_type, self._typevars_map) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2121 # Create a new ComputedFieldInfo so that different type parametrizations of the same
2122 # generic model's computed field can have different return types.
2123 d.info = dataclasses.replace(d.info, return_type=return_type) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2124 return_type_schema = self.generate_schema(return_type) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2125 # Apply serializers to computed field if there exist
2126 return_type_schema = self._apply_field_serializers( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2127 return_type_schema,
2128 filter_field_decorator_info_by_field(field_serializers.values(), d.cls_var_name),
2129 )
2131 pydantic_js_updates, pydantic_js_extra = _extract_json_schema_info_from_field_info(d.info) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2132 core_metadata: dict[str, Any] = {} 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2133 update_core_metadata( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2134 core_metadata,
2135 pydantic_js_updates={'readOnly': True, **(pydantic_js_updates if pydantic_js_updates else {})},
2136 pydantic_js_extra=pydantic_js_extra,
2137 )
2138 return core_schema.computed_field( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2139 d.cls_var_name, return_schema=return_type_schema, alias=d.info.alias, metadata=core_metadata
2140 )
2142 def _annotated_schema(self, annotated_type: Any) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2143 """Generate schema for an Annotated type, e.g. `Annotated[int, Field(...)]` or `Annotated[int, Gt(0)]`."""
2144 FieldInfo = import_cached_field_info() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2145 source_type, *annotations = self._get_args_resolving_forward_refs( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2146 annotated_type,
2147 required=True,
2148 )
2149 schema = self._apply_annotations(source_type, annotations) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2150 # put the default validator last so that TypeAdapter.get_default_value() works
2151 # even if there are function validators involved
2152 for annotation in annotations: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2153 if isinstance(annotation, FieldInfo): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2154 schema = wrap_default(annotation, schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2155 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2157 def _apply_annotations( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2158 self,
2159 source_type: Any,
2160 annotations: list[Any],
2161 transform_inner_schema: Callable[[CoreSchema], CoreSchema] = lambda x: x,
2162 check_unsupported_field_info_attributes: bool = True,
2163 ) -> CoreSchema:
2164 """Apply arguments from `Annotated` or from `FieldInfo` to a schema.
2166 This gets called by `GenerateSchema._annotated_schema` but differs from it in that it does
2167 not expect `source_type` to be an `Annotated` object, it expects it to be the first argument of that
2168 (in other words, `GenerateSchema._annotated_schema` just unpacks `Annotated`, this process it).
2169 """
2170 annotations = list(_known_annotated_metadata.expand_grouped_metadata(annotations)) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2172 pydantic_js_annotation_functions: list[GetJsonSchemaFunction] = [] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2174 def inner_handler(obj: Any) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2175 schema = self._generate_schema_from_get_schema_method(obj, source_type) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2177 if schema is None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2178 schema = self._generate_schema_inner(obj) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2180 metadata_js_function = _extract_get_pydantic_json_schema(obj) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2181 if metadata_js_function is not None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2182 metadata_schema = resolve_original_schema(schema, self.defs) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2183 if metadata_schema is not None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2184 self._add_js_function(metadata_schema, metadata_js_function) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2185 return transform_inner_schema(schema) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2187 get_inner_schema = CallbackGetCoreSchemaHandler(inner_handler, self) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2189 for annotation in annotations: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2190 if annotation is None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2191 continue 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2192 get_inner_schema = self._get_wrapped_inner_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2193 get_inner_schema,
2194 annotation,
2195 pydantic_js_annotation_functions,
2196 check_unsupported_field_info_attributes=check_unsupported_field_info_attributes,
2197 )
2199 schema = get_inner_schema(source_type) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2200 if pydantic_js_annotation_functions: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2201 core_metadata = schema.setdefault('metadata', {}) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2202 update_core_metadata(core_metadata, pydantic_js_annotation_functions=pydantic_js_annotation_functions) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2203 return _add_custom_serialization_from_json_encoders(self._config_wrapper.json_encoders, source_type, schema) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2205 def _apply_single_annotation( 1abcenpuwACDklghrsxyEFGJimdjotvzBHI
2206 self,
2207 schema: core_schema.CoreSchema,
2208 metadata: Any,
2209 check_unsupported_field_info_attributes: bool = True,
2210 ) -> core_schema.CoreSchema:
2211 FieldInfo = import_cached_field_info() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2213 if isinstance(metadata, FieldInfo): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2214 if ( 1abcenpqfklghrsimdjot
2215 check_unsupported_field_info_attributes
2216 # HACK: we don't want to emit the warning for `FieldInfo` subclasses, because FastAPI does weird manipulations
2217 # with its subclasses and their annotations:
2218 and type(metadata) is FieldInfo
2219 and (unsupported_attributes := self._get_unsupported_field_info_attributes(metadata))
2220 ):
2221 for attr, value in unsupported_attributes: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2222 warnings.warn( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2223 f'The {attr!r} attribute with value {value!r} was provided to the `Field()` function, '
2224 f'which has no effect in the context it was used. {attr!r} is field-specific metadata, '
2225 'and can only be attached to a model field using `Annotated` metadata or by assignment. '
2226 'This may have happened because an `Annotated` type alias using the `type` statement was '
2227 'used, or if the `Field()` function was attached to a single member of a union type.',
2228 category=UnsupportedFieldAttributeWarning,
2229 )
2230 for field_metadata in metadata.metadata: 2230 ↛ 2231line 2230 didn't jump to line 2231 because the loop on line 2230 never started1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2231 schema = self._apply_single_annotation(schema, field_metadata)
2233 if metadata.discriminator is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2234 schema = self._apply_discriminator_to_union(schema, metadata.discriminator) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2235 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2237 if schema['type'] == 'nullable': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2238 # for nullable schemas, metadata is automatically applied to the inner schema
2239 inner = schema.get('schema', core_schema.any_schema()) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2240 inner = self._apply_single_annotation(inner, metadata) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2241 if inner: 2241 ↛ 2243line 2241 didn't jump to line 2243 because the condition on line 2241 was always true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2242 schema['schema'] = inner 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2243 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2245 original_schema = schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2246 ref = schema.get('ref') 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2247 if ref is not None: 2247 ↛ 2248line 2247 didn't jump to line 2248 because the condition on line 2247 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2248 schema = schema.copy()
2249 new_ref = ref + f'_{repr(metadata)}'
2250 if (existing := self.defs.get_schema_from_ref(new_ref)) is not None:
2251 return existing
2252 schema['ref'] = new_ref # pyright: ignore[reportGeneralTypeIssues]
2253 elif schema['type'] == 'definition-ref': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2254 ref = schema['schema_ref'] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2255 if (referenced_schema := self.defs.get_schema_from_ref(ref)) is not None: 2255 ↛ 2262line 2255 didn't jump to line 2262 because the condition on line 2255 was always true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2256 schema = referenced_schema.copy() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2257 new_ref = ref + f'_{repr(metadata)}' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2258 if (existing := self.defs.get_schema_from_ref(new_ref)) is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2259 return existing 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2260 schema['ref'] = new_ref # pyright: ignore[reportGeneralTypeIssues] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2262 maybe_updated_schema = _known_annotated_metadata.apply_known_metadata(metadata, schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2264 if maybe_updated_schema is not None: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2265 return maybe_updated_schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2266 return original_schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2268 def _apply_single_annotation_json_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2269 self, schema: core_schema.CoreSchema, metadata: Any
2270 ) -> core_schema.CoreSchema:
2271 FieldInfo = import_cached_field_info() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2273 if isinstance(metadata, FieldInfo): 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2274 for field_metadata in metadata.metadata: 2274 ↛ 2275line 2274 didn't jump to line 2275 because the loop on line 2274 never started1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2275 schema = self._apply_single_annotation_json_schema(schema, field_metadata)
2277 pydantic_js_updates, pydantic_js_extra = _extract_json_schema_info_from_field_info(metadata) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2278 core_metadata = schema.setdefault('metadata', {}) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2279 update_core_metadata( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2280 core_metadata, pydantic_js_updates=pydantic_js_updates, pydantic_js_extra=pydantic_js_extra
2281 )
2282 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2284 def _get_unsupported_field_info_attributes(self, field_info: FieldInfo) -> list[tuple[str, Any]]: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2285 """Get the list of unsupported `FieldInfo` attributes when not directly used in `Annotated` for field annotations."""
2286 unused_metadata: list[tuple[str, Any]] = [] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2287 for unused_metadata_name, unset_value in UNSUPPORTED_STANDALONE_FIELDINFO_ATTRIBUTES: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2288 if ( 1abqf
2289 (unused_metadata_value := getattr(field_info, unused_metadata_name)) is not unset_value
2290 # `default` and `default_factory` can still be used with a type adapter, so only include them
2291 # if used with a model-like class:
2292 and (
2293 unused_metadata_name not in ('default', 'default_factory')
2294 or self.model_type_stack.get() is not None
2295 )
2296 ):
2297 # Setting `alias` will set `validation/serialization_alias` as well, so we want to avoid duplicate warnings:
2298 if ( 1abqf
2299 unused_metadata_name not in ('validation_alias', 'serialization_alias')
2300 or 'alias' not in field_info._attributes_set
2301 ):
2302 unused_metadata.append((unused_metadata_name, unused_metadata_value)) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2304 return unused_metadata 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2306 def _get_wrapped_inner_schema( 1abcenpuwACDklghrsxyEFGJimdjotvzBHI
2307 self,
2308 get_inner_schema: GetCoreSchemaHandler,
2309 annotation: Any,
2310 pydantic_js_annotation_functions: list[GetJsonSchemaFunction],
2311 check_unsupported_field_info_attributes: bool = False,
2312 ) -> CallbackGetCoreSchemaHandler:
2313 annotation_get_schema: GetCoreSchemaFunction | None = getattr(annotation, '__get_pydantic_core_schema__', None) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2315 def new_handler(source: Any) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2316 if annotation_get_schema is not None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2317 schema = annotation_get_schema(source, get_inner_schema) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2318 else:
2319 schema = get_inner_schema(source) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2320 schema = self._apply_single_annotation( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2321 schema,
2322 annotation,
2323 check_unsupported_field_info_attributes=check_unsupported_field_info_attributes,
2324 )
2325 schema = self._apply_single_annotation_json_schema(schema, annotation) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2327 metadata_js_function = _extract_get_pydantic_json_schema(annotation) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2328 if metadata_js_function is not None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2329 pydantic_js_annotation_functions.append(metadata_js_function) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2330 return schema 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2332 return CallbackGetCoreSchemaHandler(new_handler, self) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2334 def _apply_field_serializers( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2335 self,
2336 schema: core_schema.CoreSchema,
2337 serializers: list[Decorator[FieldSerializerDecoratorInfo]],
2338 ) -> core_schema.CoreSchema:
2339 """Apply field serializers to a schema."""
2340 if serializers: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2341 schema = copy(schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2342 if schema['type'] == 'definitions': 2342 ↛ 2343line 2342 didn't jump to line 2343 because the condition on line 2342 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2343 inner_schema = schema['schema']
2344 schema['schema'] = self._apply_field_serializers(inner_schema, serializers)
2345 return schema
2346 elif 'ref' in schema: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2347 schema = self.defs.create_definition_reference_schema(schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2349 # use the last serializer to make it easy to override a serializer set on a parent model
2350 serializer = serializers[-1] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2351 is_field_serializer, info_arg = inspect_field_serializer(serializer.func, serializer.info.mode) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2353 if serializer.info.return_type is not PydanticUndefined: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2354 return_type = serializer.info.return_type 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2355 else:
2356 try: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2357 # Do not pass in globals as the function could be defined in a different module.
2358 # Instead, let `get_callable_return_type` infer the globals to use, but still pass
2359 # in locals that may contain a parent/rebuild namespace:
2360 return_type = _decorators.get_callable_return_type( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2361 serializer.func, localns=self._types_namespace.locals
2362 )
2363 except NameError as e: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2364 raise PydanticUndefinedAnnotation.from_name_error(e) from e 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2366 if return_type is PydanticUndefined: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2367 return_schema = None 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2368 else:
2369 return_schema = self.generate_schema(return_type) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2371 if serializer.info.mode == 'wrap': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2372 schema['serialization'] = core_schema.wrap_serializer_function_ser_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2373 serializer.func,
2374 is_field_serializer=is_field_serializer,
2375 info_arg=info_arg,
2376 return_schema=return_schema,
2377 when_used=serializer.info.when_used,
2378 )
2379 else:
2380 assert serializer.info.mode == 'plain' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2381 schema['serialization'] = core_schema.plain_serializer_function_ser_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2382 serializer.func,
2383 is_field_serializer=is_field_serializer,
2384 info_arg=info_arg,
2385 return_schema=return_schema,
2386 when_used=serializer.info.when_used,
2387 )
2388 return schema 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2390 def _apply_model_serializers( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2391 self, schema: core_schema.CoreSchema, serializers: Iterable[Decorator[ModelSerializerDecoratorInfo]]
2392 ) -> core_schema.CoreSchema:
2393 """Apply model serializers to a schema."""
2394 ref: str | None = schema.pop('ref', None) # type: ignore 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2395 if serializers: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2396 serializer = list(serializers)[-1] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2397 info_arg = inspect_model_serializer(serializer.func, serializer.info.mode) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2399 if serializer.info.return_type is not PydanticUndefined: 2399 ↛ 2400line 2399 didn't jump to line 2400 because the condition on line 2399 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2400 return_type = serializer.info.return_type
2401 else:
2402 try: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2403 # Do not pass in globals as the function could be defined in a different module.
2404 # Instead, let `get_callable_return_type` infer the globals to use, but still pass
2405 # in locals that may contain a parent/rebuild namespace:
2406 return_type = _decorators.get_callable_return_type( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2407 serializer.func, localns=self._types_namespace.locals
2408 )
2409 except NameError as e:
2410 raise PydanticUndefinedAnnotation.from_name_error(e) from e
2412 if return_type is PydanticUndefined: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2413 return_schema = None 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2414 else:
2415 return_schema = self.generate_schema(return_type) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2417 if serializer.info.mode == 'wrap': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2418 ser_schema: core_schema.SerSchema = core_schema.wrap_serializer_function_ser_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2419 serializer.func,
2420 info_arg=info_arg,
2421 return_schema=return_schema,
2422 when_used=serializer.info.when_used,
2423 )
2424 else:
2425 # plain
2426 ser_schema = core_schema.plain_serializer_function_ser_schema( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2427 serializer.func,
2428 info_arg=info_arg,
2429 return_schema=return_schema,
2430 when_used=serializer.info.when_used,
2431 )
2432 schema['serialization'] = ser_schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2433 if ref: 2433 ↛ 2435line 2433 didn't jump to line 2435 because the condition on line 2433 was always true1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2434 schema['ref'] = ref # type: ignore 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2435 return schema 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2438_VALIDATOR_F_MATCH: Mapping[ 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2439 tuple[FieldValidatorModes, Literal['no-info', 'with-info']],
2440 Callable[[Callable[..., Any], core_schema.CoreSchema], core_schema.CoreSchema],
2441] = {
2442 ('before', 'no-info'): lambda f, schema: core_schema.no_info_before_validator_function(f, schema),
2443 ('after', 'no-info'): lambda f, schema: core_schema.no_info_after_validator_function(f, schema),
2444 ('plain', 'no-info'): lambda f, _: core_schema.no_info_plain_validator_function(f),
2445 ('wrap', 'no-info'): lambda f, schema: core_schema.no_info_wrap_validator_function(f, schema),
2446 ('before', 'with-info'): lambda f, schema: core_schema.with_info_before_validator_function(f, schema),
2447 ('after', 'with-info'): lambda f, schema: core_schema.with_info_after_validator_function(f, schema),
2448 ('plain', 'with-info'): lambda f, _: core_schema.with_info_plain_validator_function(f),
2449 ('wrap', 'with-info'): lambda f, schema: core_schema.with_info_wrap_validator_function(f, schema),
2450}
2453# TODO V3: this function is only used for deprecated decorators. It should
2454# be removed once we drop support for those.
2455def apply_validators( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2456 schema: core_schema.CoreSchema,
2457 validators: Iterable[Decorator[RootValidatorDecoratorInfo]]
2458 | Iterable[Decorator[ValidatorDecoratorInfo]]
2459 | Iterable[Decorator[FieldValidatorDecoratorInfo]],
2460) -> core_schema.CoreSchema:
2461 """Apply validators to a schema.
2463 Args:
2464 schema: The schema to apply validators on.
2465 validators: An iterable of validators.
2466 field_name: The name of the field if validators are being applied to a model field.
2468 Returns:
2469 The updated schema.
2470 """
2471 for validator in validators: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2472 info_arg = inspect_validator(validator.func, validator.info.mode) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2473 val_type = 'with-info' if info_arg else 'no-info' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2475 schema = _VALIDATOR_F_MATCH[(validator.info.mode, val_type)](validator.func, schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2476 return schema 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2479def _validators_require_validate_default(validators: Iterable[Decorator[ValidatorDecoratorInfo]]) -> bool: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2480 """In v1, if any of the validators for a field had `always=True`, the default value would be validated.
2482 This serves as an auxiliary function for re-implementing that logic, by looping over a provided
2483 collection of (v1-style) ValidatorDecoratorInfo's and checking if any of them have `always=True`.
2485 We should be able to drop this function and the associated logic calling it once we drop support
2486 for v1-style validator decorators. (Or we can extend it and keep it if we add something equivalent
2487 to the v1-validator `always` kwarg to `field_validator`.)
2488 """
2489 for validator in validators: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2490 if validator.info.always: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2491 return True 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2492 return False 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2495def apply_model_validators( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2496 schema: core_schema.CoreSchema,
2497 validators: Iterable[Decorator[ModelValidatorDecoratorInfo]],
2498 mode: Literal['inner', 'outer', 'all'],
2499) -> core_schema.CoreSchema:
2500 """Apply model validators to a schema.
2502 If mode == 'inner', only "before" validators are applied
2503 If mode == 'outer', validators other than "before" are applied
2504 If mode == 'all', all validators are applied
2506 Args:
2507 schema: The schema to apply validators on.
2508 validators: An iterable of validators.
2509 mode: The validator mode.
2511 Returns:
2512 The updated schema.
2513 """
2514 ref: str | None = schema.pop('ref', None) # type: ignore 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2515 for validator in validators: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2516 if mode == 'inner' and validator.info.mode != 'before': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2517 continue 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2518 if mode == 'outer' and validator.info.mode == 'before': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2519 continue 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2520 info_arg = inspect_validator(validator.func, validator.info.mode) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2521 if validator.info.mode == 'wrap': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2522 if info_arg: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2523 schema = core_schema.with_info_wrap_validator_function(function=validator.func, schema=schema) 1uwACDxyEFGvzBHI
2524 else:
2525 schema = core_schema.no_info_wrap_validator_function(function=validator.func, schema=schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2526 elif validator.info.mode == 'before': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2527 if info_arg: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2528 schema = core_schema.with_info_before_validator_function(function=validator.func, schema=schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2529 else:
2530 schema = core_schema.no_info_before_validator_function(function=validator.func, schema=schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2531 else:
2532 assert validator.info.mode == 'after' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2533 if info_arg: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2534 schema = core_schema.with_info_after_validator_function(function=validator.func, schema=schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2535 else:
2536 schema = core_schema.no_info_after_validator_function(function=validator.func, schema=schema) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2537 if ref: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2538 schema['ref'] = ref # type: ignore 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2539 return schema 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2542def wrap_default(field_info: FieldInfo, schema: core_schema.CoreSchema) -> core_schema.CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2543 """Wrap schema with default schema if default value or `default_factory` are available.
2545 Args:
2546 field_info: The field info object.
2547 schema: The schema to apply default on.
2549 Returns:
2550 Updated schema by default value or `default_factory`.
2551 """
2552 if field_info.default_factory: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2553 return core_schema.with_default_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2554 schema,
2555 default_factory=field_info.default_factory,
2556 default_factory_takes_data=takes_validated_data_argument(field_info.default_factory),
2557 validate_default=field_info.validate_default,
2558 )
2559 elif field_info.default is not PydanticUndefined: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2560 return core_schema.with_default_schema( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2561 schema, default=field_info.default, validate_default=field_info.validate_default
2562 )
2563 else:
2564 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2567def _extract_get_pydantic_json_schema(tp: Any) -> GetJsonSchemaFunction | None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2568 """Extract `__get_pydantic_json_schema__` from a type, handling the deprecated `__modify_schema__`."""
2569 js_modify_function = getattr(tp, '__get_pydantic_json_schema__', None) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2571 if hasattr(tp, '__modify_schema__'): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2572 BaseModel = import_cached_base_model() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2574 has_custom_v2_modify_js_func = ( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2575 js_modify_function is not None
2576 and BaseModel.__get_pydantic_json_schema__.__func__ # type: ignore
2577 not in (js_modify_function, getattr(js_modify_function, '__func__', None))
2578 )
2580 if not has_custom_v2_modify_js_func: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2581 cls_name = getattr(tp, '__name__', None) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2582 raise PydanticUserError( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2583 f'The `__modify_schema__` method is not supported in Pydantic v2. '
2584 f'Use `__get_pydantic_json_schema__` instead{f" in class `{cls_name}`" if cls_name else ""}.',
2585 code='custom-json-schema',
2586 )
2588 if (origin := get_origin(tp)) is not None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2589 # Generic aliases proxy attribute access to the origin, *except* dunder attributes,
2590 # such as `__get_pydantic_json_schema__`, hence the explicit check.
2591 return _extract_get_pydantic_json_schema(origin) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2593 if js_modify_function is None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2594 return None 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2596 return js_modify_function 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2599class _CommonField(TypedDict): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2600 schema: core_schema.CoreSchema 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2601 validation_alias: str | list[str | int] | list[list[str | int]] | None 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2602 serialization_alias: str | None 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2603 serialization_exclude: bool | None 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2604 frozen: bool | None 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2605 metadata: dict[str, Any] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2608def _common_field( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2609 schema: core_schema.CoreSchema,
2610 *,
2611 validation_alias: str | list[str | int] | list[list[str | int]] | None = None,
2612 serialization_alias: str | None = None,
2613 serialization_exclude: bool | None = None,
2614 frozen: bool | None = None,
2615 metadata: Any = None,
2616) -> _CommonField:
2617 return { 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2618 'schema': schema,
2619 'validation_alias': validation_alias,
2620 'serialization_alias': serialization_alias,
2621 'serialization_exclude': serialization_exclude,
2622 'frozen': frozen,
2623 'metadata': metadata,
2624 }
2627def resolve_original_schema(schema: CoreSchema, definitions: _Definitions) -> CoreSchema | None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2628 if schema['type'] == 'definition-ref': 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2629 return definitions.get_schema_from_ref(schema['schema_ref']) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2630 elif schema['type'] == 'definitions': 2630 ↛ 2631line 2630 didn't jump to line 2631 because the condition on line 2630 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2631 return schema['schema']
2632 else:
2633 return schema 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2636def _inlining_behavior( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2637 def_ref: core_schema.DefinitionReferenceSchema,
2638) -> Literal['inline', 'keep', 'preserve_metadata']:
2639 """Determine the inlining behavior of the `'definition-ref'` schema.
2641 - If no `'serialization'` schema and no metadata is attached, the schema can safely be inlined.
2642 - If it has metadata but only related to the deferred discriminator application, it can be inlined
2643 provided that such metadata is kept.
2644 - Otherwise, the schema should not be inlined. Doing so would remove the `'serialization'` schema or metadata.
2645 """
2646 if 'serialization' in def_ref: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2647 return 'keep' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2648 metadata = def_ref.get('metadata') 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2649 if not metadata: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2650 return 'inline' 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2651 if len(metadata) == 1 and 'pydantic_internal_union_discriminator' in metadata: 2651 ↛ 2652line 2651 didn't jump to line 2652 because the condition on line 2651 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2652 return 'preserve_metadata'
2653 return 'keep' 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2656class _Definitions: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2657 """Keeps track of references and definitions."""
2659 _recursively_seen: set[str] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2660 """A set of recursively seen references. 1cenpuwACDfghrsxyEFGJdjotvzBHI
2662 When a referenceable type is encountered, the `get_schema_or_ref` context manager is
2663 entered to compute the reference. If the type references itself by some way (e.g. for
2664 a dataclass a Pydantic model, the class can be referenced as a field annotation),
2665 entering the context manager again will yield a `'definition-ref'` schema that should
2666 short-circuit the normal generation process, as the reference was already in this set.
2667 """
2669 _definitions: dict[str, core_schema.CoreSchema] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2670 """A mapping of references to their corresponding schema. 1cenpuwACDfghrsxyEFGJdjotvzBHI
2672 When a schema for a referenceable type is generated, it is stored in this mapping. If the
2673 same type is encountered again, the reference is yielded by the `get_schema_or_ref` context
2674 manager.
2675 """
2677 def __init__(self) -> None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2678 self._recursively_seen = set() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2679 self._definitions = {} 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2681 @contextmanager 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2682 def get_schema_or_ref(self, tp: Any, /) -> Generator[tuple[str, core_schema.DefinitionReferenceSchema | None]]: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2683 """Get a definition for `tp` if one exists.
2685 If a definition exists, a tuple of `(ref_string, CoreSchema)` is returned.
2686 If no definition exists yet, a tuple of `(ref_string, None)` is returned.
2688 Note that the returned `CoreSchema` will always be a `DefinitionReferenceSchema`,
2689 not the actual definition itself.
2691 This should be called for any type that can be identified by reference.
2692 This includes any recursive types.
2694 At present the following types can be named/recursive:
2696 - Pydantic model
2697 - Pydantic and stdlib dataclasses
2698 - Typed dictionaries
2699 - Named tuples
2700 - `TypeAliasType` instances
2701 - Enums
2702 """
2703 ref = get_type_ref(tp) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2704 # return the reference if we're either (1) in a cycle or (2) it the reference was already encountered:
2705 if ref in self._recursively_seen or ref in self._definitions: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2706 yield (ref, core_schema.definition_reference_schema(ref)) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2707 else:
2708 self._recursively_seen.add(ref) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2709 try: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2710 yield (ref, None) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2711 finally:
2712 self._recursively_seen.discard(ref) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2714 def get_schema_from_ref(self, ref: str) -> CoreSchema | None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2715 """Resolve the schema from the given reference."""
2716 return self._definitions.get(ref) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2718 def create_definition_reference_schema(self, schema: CoreSchema) -> core_schema.DefinitionReferenceSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2719 """Store the schema as a definition and return a `'definition-reference'` schema pointing to it.
2721 The schema must have a reference attached to it.
2722 """
2723 ref = schema['ref'] # pyright: ignore 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2724 self._definitions[ref] = schema 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2725 return core_schema.definition_reference_schema(ref) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2727 def unpack_definitions(self, schema: core_schema.DefinitionsSchema) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2728 """Store the definitions of the `'definitions'` core schema and return the inner core schema."""
2729 for def_schema in schema['definitions']: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2730 self._definitions[def_schema['ref']] = def_schema # pyright: ignore 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2731 return schema['schema'] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2733 def finalize_schema(self, schema: CoreSchema) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2734 """Finalize the core schema.
2736 This traverses the core schema and referenced definitions, replaces `'definition-ref'` schemas
2737 by the referenced definition if possible, and applies deferred discriminators.
2738 """
2739 definitions = self._definitions 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2740 try: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2741 gather_result = gather_schemas_for_cleaning( 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2742 schema,
2743 definitions=definitions,
2744 )
2745 except MissingDefinitionError as e: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2746 raise InvalidSchemaError from e 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2748 remaining_defs: dict[str, CoreSchema] = {} 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2750 # Note: this logic doesn't play well when core schemas with deferred discriminator metadata
2751 # and references are encountered. See the `test_deferred_discriminated_union_and_references()` test.
2752 for ref, inlinable_def_ref in gather_result['collected_references'].items(): 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2753 if inlinable_def_ref is not None and (inlining_behavior := _inlining_behavior(inlinable_def_ref)) != 'keep': 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2754 if inlining_behavior == 'inline': 2754 ↛ 2761line 2754 didn't jump to line 2761 because the condition on line 2754 was always true1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2755 # `ref` was encountered, and only once:
2756 # - `inlinable_def_ref` is a `'definition-ref'` schema and is guaranteed to be
2757 # the only one. Transform it into the definition it points to.
2758 # - Do not store the definition in the `remaining_defs`.
2759 inlinable_def_ref.clear() # pyright: ignore[reportAttributeAccessIssue] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2760 inlinable_def_ref.update(self._resolve_definition(ref, definitions)) # pyright: ignore 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2761 elif inlining_behavior == 'preserve_metadata':
2762 # `ref` was encountered, and only once, but contains discriminator metadata.
2763 # We will do the same thing as if `inlining_behavior` was `'inline'`, but make
2764 # sure to keep the metadata for the deferred discriminator application logic below.
2765 meta = inlinable_def_ref.pop('metadata')
2766 inlinable_def_ref.clear() # pyright: ignore[reportAttributeAccessIssue]
2767 inlinable_def_ref.update(self._resolve_definition(ref, definitions)) # pyright: ignore
2768 inlinable_def_ref['metadata'] = meta
2769 else:
2770 # `ref` was encountered, at least two times (or only once, but with metadata or a serialization schema):
2771 # - Do not inline the `'definition-ref'` schemas (they are not provided in the gather result anyway).
2772 # - Store the the definition in the `remaining_defs`
2773 remaining_defs[ref] = self._resolve_definition(ref, definitions) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2775 for cs in gather_result['deferred_discriminator_schemas']: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2776 discriminator: str | None = cs['metadata'].pop('pydantic_internal_union_discriminator', None) # pyright: ignore[reportTypedDictNotRequiredAccess] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2777 if discriminator is None: 2777 ↛ 2781line 2777 didn't jump to line 2781 because the condition on line 2777 was never true1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2778 # This can happen in rare scenarios, when a deferred schema is present multiple times in the
2779 # gather result (e.g. when using the `Sequence` type -- see `test_sequence_discriminated_union()`).
2780 # In this case, a previous loop iteration applied the discriminator and so we can just skip it here.
2781 continue
2782 applied = _discriminated_union.apply_discriminator(cs.copy(), discriminator, remaining_defs) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2783 # Mutate the schema directly to have the discriminator applied
2784 cs.clear() # pyright: ignore[reportAttributeAccessIssue] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2785 cs.update(applied) # pyright: ignore 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2787 if remaining_defs: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2788 schema = core_schema.definitions_schema(schema=schema, definitions=[*remaining_defs.values()]) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2789 return schema 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2791 def _resolve_definition(self, ref: str, definitions: dict[str, CoreSchema]) -> CoreSchema: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2792 definition = definitions[ref] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2793 if definition['type'] != 'definition-ref': 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2794 return definition 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2796 # Some `'definition-ref'` schemas might act as "intermediate" references (e.g. when using
2797 # a PEP 695 type alias (which is referenceable) that references another PEP 695 type alias):
2798 visited: set[str] = set() 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2799 while definition['type'] == 'definition-ref' and _inlining_behavior(definition) == 'inline': 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2800 schema_ref = definition['schema_ref'] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2801 if schema_ref in visited: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2802 raise PydanticUserError( 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2803 f'{ref} contains a circular reference to itself.', code='circular-reference-schema'
2804 )
2805 visited.add(schema_ref) 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2806 definition = definitions[schema_ref] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2807 return {**definition, 'ref': ref} # pyright: ignore[reportReturnType] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2810class _FieldNameStack: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2811 __slots__ = ('_stack',) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2813 def __init__(self) -> None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2814 self._stack: list[str] = [] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2816 @contextmanager 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2817 def push(self, field_name: str) -> Iterator[None]: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2818 self._stack.append(field_name) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2819 yield 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2820 self._stack.pop() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2822 def get(self) -> str | None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2823 if self._stack:
2824 return self._stack[-1]
2825 else:
2826 return None
2829class _ModelTypeStack: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2830 __slots__ = ('_stack',) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2832 def __init__(self) -> None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2833 self._stack: list[type] = [] 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2835 @contextmanager 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2836 def push(self, type_obj: type) -> Iterator[None]: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2837 self._stack.append(type_obj) 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2838 yield 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2839 self._stack.pop() 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2841 def get(self) -> type | None: 1abcenpuwACDqfklghrsxyEFGJimdjotvzBHI
2842 if self._stack: 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2843 return self._stack[-1] 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI
2844 else:
2845 return None 1abcenpuwACDqfklghrsxyEFGimdjotvzBHI