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