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