Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Convert python types to pydantic-core schema."""
3from __future__ import annotations as _annotations
5import collections.abc
6import dataclasses
7import datetime
8import inspect
9import os
10import pathlib
11import re
12import sys
13import typing
14import warnings
15from collections.abc import Callable, Generator, Iterable, Iterator, Mapping
16from contextlib import contextmanager
17from copy import copy
18from decimal import Decimal
19from enum import Enum
20from fractions import Fraction
21from functools import partial
22from inspect import Parameter, _ParameterKind
23from ipaddress import IPv4Address, IPv4Interface, IPv4Network, IPv6Address, IPv6Interface, IPv6Network
24from itertools import chain
25from operator import attrgetter
26from types import EllipsisType, FunctionType, GenericAlias, LambdaType, MethodType, NoneType
27from typing import (
28 TYPE_CHECKING,
29 Any,
30 ClassVar,
31 ForwardRef,
32 Literal,
33 TypeAlias,
34 TypeVar,
35 cast,
36 overload,
37)
38from uuid import UUID
39from zoneinfo import ZoneInfo
41import typing_extensions
42from pydantic_core import (
43 MISSING,
44 CoreSchema,
45 MultiHostUrl,
46 PydanticCustomError,
47 PydanticSerializationUnexpectedValue,
48 PydanticUndefined,
49 Url,
50 core_schema,
51 to_jsonable_python,
52)
53from typing_extensions import ( # noqa: UP035 (for `get_args` and `get_origin`)
54 TypeAliasType,
55 get_args,
56 get_origin,
57 is_typeddict,
58)
59from typing_inspection import typing_objects
60from typing_inspection.introspection import AnnotationSource, get_literal_values, is_union_origin
62from ..aliases import AliasChoices, AliasPath
63from ..annotated_handlers import GetCoreSchemaHandler, GetJsonSchemaHandler
64from ..config import ConfigDict, JsonDict, JsonEncoder, JsonSchemaExtraCallable
65from ..errors import (
66 PydanticForbiddenQualifier,
67 PydanticInvalidForJsonSchema,
68 PydanticSchemaGenerationError,
69 PydanticUndefinedAnnotation,
70 PydanticUserError,
71)
72from ..functional_validators import AfterValidator, BeforeValidator, FieldValidatorModes, PlainValidator, WrapValidator
73from ..json_schema import JsonSchemaValue
74from ..version import version_short
75from ..warnings import (
76 ArbitraryTypeWarning,
77 PydanticDeprecatedSince20,
78 TypedDictExtraConfigWarning,
79 UnsupportedFieldAttributeWarning,
80)
81from . import _decorators, _discriminated_union, _known_annotated_metadata, _repr, _type_refs, _typing_extra
82from ._config import ConfigWrapper, ConfigWrapperStack
83from ._core_metadata import CoreMetadata, update_core_metadata
84from ._core_utils import (
85 get_ref,
86 is_list_like_schema_with_items_schema,
87)
88from ._decorators import (
89 Decorator,
90 DecoratorInfos,
91 FieldSerializerDecoratorInfo,
92 FieldValidatorDecoratorInfo,
93 ModelSerializerDecoratorInfo,
94 ModelValidatorDecoratorInfo,
95 RootValidatorDecoratorInfo,
96 ValidatorDecoratorInfo,
97 get_attribute_from_bases,
98 inspect_field_serializer,
99 inspect_model_serializer,
100 inspect_validator,
101 mro,
102)
103from ._docs_extraction import extract_docstrings_from_cls
104from ._fields import (
105 collect_dataclass_fields,
106 rebuild_dataclass_fields,
107 rebuild_model_fields,
108 takes_validated_data_argument,
109 update_field_from_config,
110)
111from ._forward_ref import PydanticRecursiveRef
112from ._generics import get_standard_typevars_map, replace_types
113from ._import_utils import import_cached_base_model, import_cached_field_info
114from ._mock_val_ser import MockCoreSchema
115from ._namespace_utils import NamespacesTuple, NsResolver
116from ._schema_gather import MissingDefinitionError, gather_schemas_for_cleaning
117from ._schema_generation_shared import CallbackGetCoreSchemaHandler
118from ._utils import lenient_issubclass, smart_deepcopy
120if TYPE_CHECKING:
121 from ..fields import ComputedFieldInfo, FieldInfo
122 from ..main import BaseModel
123 from ..types import Discriminator
124 from ._dataclasses import StandardDataclass
125 from ._schema_generation_shared import GetJsonSchemaFunction
127if sys.version_info >= (3, 15):
128 from builtins import frozendict
130_SUPPORTS_TYPEDDICT = sys.version_info >= (3, 12)
132FieldDecoratorInfo: TypeAlias = ValidatorDecoratorInfo | FieldValidatorDecoratorInfo | FieldSerializerDecoratorInfo
133FieldDecoratorInfoType = TypeVar('FieldDecoratorInfoType', bound=FieldDecoratorInfo)
134AnyFieldDecorator = (
135 Decorator[ValidatorDecoratorInfo] | Decorator[FieldValidatorDecoratorInfo] | Decorator[FieldSerializerDecoratorInfo]
136)
138ModifyCoreSchemaWrapHandler: TypeAlias = GetCoreSchemaHandler
139GetCoreSchemaFunction: TypeAlias = Callable[[Any, ModifyCoreSchemaWrapHandler], core_schema.CoreSchema]
140ParametersCallback: TypeAlias = Callable[[int, str, Any], Literal['skip'] | None]
142TUPLE_TYPES: list[type] = [typing.Tuple, tuple] # noqa: UP006
144# Note: This does not play very well with type checkers. For example,
145# `a: LambdaType = lambda x: x` will raise a type error by Pyright.
146ValidateCallSupportedTypes: TypeAlias = LambdaType | FunctionType | MethodType | partial
149VALIDATE_CALL_SUPPORTED_TYPES = get_args(ValidateCallSupportedTypes)
150UNSUPPORTED_STANDALONE_FIELDINFO_ATTRIBUTES: list[tuple[str, Any]] = [
151 ('alias', None),
152 ('validation_alias', None),
153 ('serialization_alias', None),
154 # will be set if any alias is set, so disable it to avoid double warnings:
155 # 'alias_priority',
156 ('default', PydanticUndefined),
157 ('default_factory', None),
158 ('exclude', None),
159 ('deprecated', None),
160 ('repr', True),
161 ('validate_default', None),
162 ('frozen', None),
163 ('init', None),
164 ('init_var', None),
165 ('kw_only', None),
166]
167"""`FieldInfo` attributes (and their default value) that can't be used outside of a model (e.g. in a type adapter or a PEP 695 type alias)."""
169_mode_to_validator: dict[
170 FieldValidatorModes, type[BeforeValidator | AfterValidator | PlainValidator | WrapValidator]
171] = {'before': BeforeValidator, 'after': AfterValidator, 'plain': PlainValidator, 'wrap': WrapValidator}
174def check_validator_fields_against_field_name(
175 info: FieldDecoratorInfo,
176 field: str,
177) -> bool:
178 """Check if field name is in validator fields.
180 Args:
181 info: The field info.
182 field: The field name to check.
184 Returns:
185 `True` if field name is in validator fields, `False` otherwise.
186 """
187 fields = info.fields
188 return '*' in fields or field in fields
191def check_decorator_fields_exist(decorators: Iterable[AnyFieldDecorator], fields: Iterable[str]) -> None:
192 """Check if the defined fields in decorators exist in `fields` param.
194 It ignores the check for a decorator if the decorator has `*` as field or `check_fields=False`.
196 Args:
197 decorators: An iterable of decorators.
198 fields: An iterable of fields name.
200 Raises:
201 PydanticUserError: If one of the field names does not exist in `fields` param.
202 """
203 fields = set(fields)
204 for dec in decorators:
205 if '*' in dec.info.fields:
206 continue
207 if dec.info.check_fields is False:
208 continue
209 for field in dec.info.fields:
210 if field not in fields:
211 raise PydanticUserError(
212 f'Decorators defined with incorrect fields: {dec.cls_ref}.{dec.cls_var_name}'
213 " (use check_fields=False if you're inheriting from the model and intended this)",
214 code='decorator-missing-field',
215 )
218def filter_field_decorator_info_by_field(
219 validator_functions: Iterable[Decorator[FieldDecoratorInfoType]], field: str
220) -> list[Decorator[FieldDecoratorInfoType]]:
221 return [dec for dec in validator_functions if check_validator_fields_against_field_name(dec.info, field)]
224def apply_each_item_validators(
225 schema: core_schema.CoreSchema,
226 each_item_validators: list[Decorator[ValidatorDecoratorInfo]],
227) -> core_schema.CoreSchema:
228 # This V1 compatibility shim should eventually be removed
230 # fail early if each_item_validators is empty
231 if not each_item_validators:
232 return schema
234 # push down any `each_item=True` validators
235 # note that this won't work for any Annotated types that get wrapped by a function validator
236 # but that's okay because that didn't exist in V1
237 if schema['type'] == 'nullable':
238 schema['schema'] = apply_each_item_validators(schema['schema'], each_item_validators)
239 return schema
240 elif schema['type'] == 'missing-sentinel' and 'schema' in schema:
241 schema['schema'] = apply_each_item_validators(schema['schema'], each_item_validators)
242 return schema
243 elif schema['type'] == 'tuple':
244 if (variadic_item_index := schema.get('variadic_item_index')) is not None:
245 schema['items_schema'][variadic_item_index] = apply_validators(
246 schema['items_schema'][variadic_item_index],
247 each_item_validators,
248 )
249 elif is_list_like_schema_with_items_schema(schema):
250 inner_schema = schema.get('items_schema', core_schema.any_schema())
251 schema['items_schema'] = apply_validators(inner_schema, each_item_validators)
252 elif schema['type'] == 'dict':
253 inner_schema = schema.get('values_schema', core_schema.any_schema())
254 schema['values_schema'] = apply_validators(inner_schema, each_item_validators)
255 else:
256 raise TypeError(
257 f'`@validator(..., each_item=True)` cannot be applied to fields with a schema of {schema["type"]}'
258 )
259 return schema
262def _extract_json_schema_info_from_field_info(
263 info: FieldInfo | ComputedFieldInfo,
264) -> tuple[JsonDict | None, JsonDict | JsonSchemaExtraCallable | None]:
265 json_schema_updates = {
266 'title': info.title,
267 'description': info.description,
268 'deprecated': bool(info.deprecated) or info.deprecated == '' or None,
269 'examples': to_jsonable_python(info.examples),
270 }
271 json_schema_updates = {k: v for k, v in json_schema_updates.items() if v is not None}
272 return (json_schema_updates or None, info.json_schema_extra)
275JsonEncoders = dict[type[Any], JsonEncoder]
278def _add_custom_serialization_from_json_encoders(
279 json_encoders: JsonEncoders | None, tp: Any, schema: CoreSchema
280) -> CoreSchema:
281 """Iterate over the json_encoders and add the first matching encoder to the schema.
283 Args:
284 json_encoders: A dictionary of types and their encoder functions.
285 tp: The type to check for a matching encoder.
286 schema: The schema to add the encoder to.
287 """
288 if not json_encoders:
289 return schema
290 if 'serialization' in schema:
291 return schema
292 # Check the class type and its superclasses for a matching encoder
293 # Decimal.__class__.__mro__ (and probably other cases) doesn't include Decimal itself
294 # if the type is a GenericAlias (e.g. from list[int]) we need to use __class__ instead of .__mro__
295 for base in (tp, *getattr(tp, '__mro__', tp.__class__.__mro__)[:-1]):
296 encoder = json_encoders.get(base)
297 if encoder is None:
298 continue
300 warnings.warn(
301 f'`json_encoders` is deprecated. See https://pydantic.dev/docs/validation/{version_short()}/concepts/serialization/#custom-serializers for alternatives',
302 PydanticDeprecatedSince20,
303 )
305 # TODO: in theory we should check that the schema accepts a serialization key
306 schema['serialization'] = core_schema.plain_serializer_function_ser_schema(encoder, when_used='json')
307 return schema
309 return schema
312GENERATE_SCHEMA_ERRORS = (
313 PydanticForbiddenQualifier,
314 PydanticInvalidForJsonSchema,
315 PydanticSchemaGenerationError,
316 PydanticUndefinedAnnotation,
317)
318"""Errors raised during core schema generation. This does *not* include `InvalidSchemaError`, which is raised during schema cleaning."""
321def _get_typed_dict_extra_items(typed_dict_cls: type[Any]) -> tuple[bool, Any, bool]:
322 """Return whether the TypedDict class is closed, the type of its extra items and whether
323 they are read-only (as per PEP 728).
325 The `__closed__` and `__extra_items__` attributes only reflect the arguments passed to the class itself,
326 so the first class in the (virtual) MRO explicitly specifying one of them is used.
327 `extra_items=Never` is normalized to `closed=True`, and the `ReadOnly` qualifier is stripped.
328 """
329 for base in mro(typed_dict_cls):
330 # `__closed__` is `None` when not specified:
331 closed = base.__dict__.get('__closed__')
332 extra_items = base.__dict__.get('__extra_items__', typing_extensions.NoExtraItems)
333 if closed is not None or not typing_objects.is_noextraitems(extra_items):
334 break
335 else:
336 return False, typing_extensions.NoExtraItems, False
338 if closed is not None:
339 # `closed` and `extra_items` can't be used together:
340 return closed, typing_extensions.NoExtraItems, False
342 readonly = typing_objects.is_readonly(get_origin(extra_items))
343 if readonly:
344 extra_items = get_args(extra_items)[0]
345 # TODO: `Never` wrapped in a type alias (e.g. `type NoExtras = Never`) or in `Annotated` isn't detected here,
346 # and fails during schema generation. Unwrapping should only apply to this check (to preserve alias handling).
347 if typing_objects.is_never(extra_items):
348 return True, typing_extensions.NoExtraItems, False
349 return False, extra_items, readonly
352class InvalidSchemaError(Exception):
353 """The core schema is invalid."""
356class GenerateSchema:
357 """Generate core schema for a Pydantic model, dataclass and types like `str`, `datetime`, ... ."""
359 __slots__ = (
360 '_config_wrapper_stack',
361 '_ns_resolver',
362 '_typevars_map',
363 'field_name_stack',
364 'model_type_stack',
365 'defs',
366 )
368 def __init__(
369 self,
370 config_wrapper: ConfigWrapper,
371 ns_resolver: NsResolver | None = None,
372 typevars_map: Mapping[TypeVar, Any] | None = None,
373 ) -> None:
374 # we need a stack for recursing into nested models
375 self._config_wrapper_stack = ConfigWrapperStack(config_wrapper)
376 self._ns_resolver = ns_resolver or NsResolver()
377 self._typevars_map = typevars_map
378 self.field_name_stack = _FieldNameStack()
379 self.model_type_stack = _ModelTypeStack()
380 self.defs = _Definitions()
382 # The handler lambdas take the `GenerateSchema` instance as their first argument, so that
383 # the dicts can be built once at class creation time and shared by all instances:
384 _handlers: ClassVar[dict[Any, Callable[[GenerateSchema, Any], core_schema.CoreSchema]]] = {
385 str: lambda self, obj: core_schema.str_schema(),
386 bytes: lambda self, obj: core_schema.bytes_schema(),
387 int: lambda self, obj: core_schema.int_schema(),
388 float: lambda self, obj: core_schema.float_schema(),
389 bool: lambda self, obj: core_schema.bool_schema(),
390 complex: lambda self, obj: core_schema.complex_schema(),
391 object: lambda self, obj: core_schema.any_schema(),
392 datetime.date: lambda self, obj: core_schema.date_schema(),
393 datetime.datetime: lambda self, obj: core_schema.datetime_schema(),
394 datetime.time: lambda self, obj: core_schema.time_schema(),
395 datetime.timedelta: lambda self, obj: core_schema.timedelta_schema(),
396 Decimal: lambda self, obj: core_schema.decimal_schema(),
397 UUID: lambda self, obj: core_schema.uuid_schema(),
398 Url: lambda self, obj: core_schema.url_schema(),
399 Fraction: lambda self, obj: core_schema.fraction_schema(),
400 MultiHostUrl: lambda self, obj: core_schema.multi_host_url_schema(),
401 None: lambda self, obj: core_schema.none_schema(),
402 NoneType: lambda self, obj: core_schema.none_schema(),
403 MISSING: lambda self, obj: core_schema.missing_sentinel_schema(),
404 EllipsisType: lambda self, obj: core_schema.ellipsis_schema(),
405 type: lambda self, obj: self._type_schema(),
406 dict: lambda self, obj: self._dict_schema(Any, Any),
407 tuple: lambda self, obj: self._tuple_schema(obj),
408 list: lambda self, obj: self._list_schema(Any),
409 set: lambda self, obj: self._set_schema(Any),
410 collections.abc.MutableSet: lambda self, obj: self._set_schema(Any),
411 frozenset: lambda self, obj: self._frozenset_schema(Any),
412 collections.abc.Set: lambda self, obj: self._frozenset_schema(Any),
413 collections.abc.MutableSequence: lambda self, obj: self._list_schema(Any),
414 collections.abc.Sequence: lambda self, obj: self._sequence_schema(Any),
415 collections.deque: lambda self, obj: self._deque_schema(Any),
416 collections.abc.Iterable: lambda self, obj: self._iterable_schema(obj),
417 collections.abc.Generator: lambda self, obj: self._iterable_schema(obj),
418 collections.abc.Mapping: lambda self, obj: self._mapping_schema(obj, Any, Any),
419 collections.abc.MutableMapping: lambda self, obj: self._mapping_schema(obj, Any, Any),
420 collections.OrderedDict: lambda self, obj: self._ordered_dict_schema(Any, Any),
421 collections.defaultdict: lambda self, obj: self._mapping_schema(obj, Any, Any),
422 collections.Counter: lambda self, obj: self._counter_schema(Any),
423 collections.abc.Callable: lambda self, obj: core_schema.callable_schema(),
424 collections.abc.Hashable: lambda self, obj: self._hashable_schema(),
425 IPv4Address: lambda self, obj: self._ip_schema(obj),
426 IPv4Interface: lambda self, obj: self._ip_schema(obj),
427 IPv4Network: lambda self, obj: self._ip_schema(obj),
428 IPv6Address: lambda self, obj: self._ip_schema(obj),
429 IPv6Interface: lambda self, obj: self._ip_schema(obj),
430 IPv6Network: lambda self, obj: self._ip_schema(obj),
431 os.PathLike: lambda self, obj: self._path_schema(obj, Any),
432 pathlib.Path: lambda self, obj: self._path_schema(obj, Any),
433 pathlib.PurePath: lambda self, obj: self._path_schema(obj, Any),
434 pathlib.PosixPath: lambda self, obj: self._path_schema(obj, Any),
435 pathlib.WindowsPath: lambda self, obj: self._path_schema(obj, Any),
436 pathlib.PurePosixPath: lambda self, obj: self._path_schema(obj, Any),
437 pathlib.PureWindowsPath: lambda self, obj: self._path_schema(obj, Any),
438 re.Pattern: lambda self, obj: self._pattern_schema(obj),
439 ZoneInfo: lambda self, obj: self._zoneinfo_schema(),
440 }
442 _generic_handlers: ClassVar[dict[Any, Callable[[GenerateSchema, Any], core_schema.CoreSchema]]] = {
443 tuple: lambda self, obj: self._tuple_schema(obj),
444 list: lambda self, obj: self._list_schema(self._get_first_arg_or_any(obj)),
445 dict: lambda self, obj: self._dict_schema(*self._get_first_two_args_or_any(obj)),
446 set: lambda self, obj: self._set_schema(self._get_first_arg_or_any(obj)),
447 frozenset: lambda self, obj: self._frozenset_schema(self._get_first_arg_or_any(obj)),
448 type: lambda self, obj: self._subclass_schema(obj),
449 collections.abc.MutableSequence: lambda self, obj: self._list_schema(self._get_first_arg_or_any(obj)),
450 collections.abc.MutableSet: lambda self, obj: self._set_schema(self._get_first_arg_or_any(obj)),
451 collections.abc.Set: lambda self, obj: self._frozenset_schema(self._get_first_arg_or_any(obj)),
452 collections.deque: lambda self, obj: self._deque_schema(self._get_first_arg_or_any(obj)),
453 collections.abc.Mapping: lambda self, obj: self._mapping_schema(
454 collections.abc.Mapping, *self._get_first_two_args_or_any(obj)
455 ),
456 collections.abc.MutableMapping: lambda self, obj: self._mapping_schema(
457 collections.abc.MutableMapping, *self._get_first_two_args_or_any(obj)
458 ),
459 collections.OrderedDict: lambda self, obj: self._ordered_dict_schema(*self._get_first_two_args_or_any(obj)),
460 collections.defaultdict: lambda self, obj: self._mapping_schema(
461 collections.defaultdict, *self._get_first_two_args_or_any(obj)
462 ),
463 collections.Counter: lambda self, obj: self._counter_schema(self._get_first_arg_or_any(obj)),
464 os.PathLike: lambda self, obj: self._path_schema(os.PathLike, self._get_first_arg_or_any(obj)),
465 pathlib.Path: lambda self, obj: self._path_schema(pathlib.Path, self._get_first_arg_or_any(obj)),
466 pathlib.PurePath: lambda self, obj: self._path_schema(pathlib.PurePath, self._get_first_arg_or_any(obj)),
467 pathlib.PosixPath: lambda self, obj: self._path_schema(pathlib.PosixPath, self._get_first_arg_or_any(obj)),
468 pathlib.WindowsPath: lambda self, obj: self._path_schema(pathlib.WindowsPath, self._get_first_arg_or_any(obj)),
469 pathlib.PurePosixPath: lambda self, obj: self._path_schema(
470 pathlib.PurePosixPath, self._get_first_arg_or_any(obj)
471 ),
472 pathlib.PureWindowsPath: lambda self, obj: self._path_schema(
473 pathlib.PureWindowsPath, self._get_first_arg_or_any(obj)
474 ),
475 collections.abc.Sequence: lambda self, obj: self._sequence_schema(self._get_first_arg_or_any(obj)),
476 collections.abc.Iterable: lambda self, obj: self._iterable_schema(obj),
477 collections.abc.Generator: lambda self, obj: self._iterable_schema(obj),
478 re.Pattern: lambda self, obj: self._pattern_schema(obj),
479 collections.abc.Callable: lambda self, obj: core_schema.callable_schema(),
480 }
482 if sys.version_info >= (3, 15):
483 _handlers[frozendict] = lambda self, obj: self._frozendict_schema(Any, Any)
484 _generic_handlers[frozendict] = lambda self, obj: self._frozendict_schema(*self._get_first_two_args_or_any(obj))
486 def __init_subclass__(cls) -> None:
487 super().__init_subclass__()
488 warnings.warn(
489 'Subclassing `GenerateSchema` is not supported. The API is highly subject to change in minor versions.',
490 UserWarning,
491 stacklevel=2,
492 )
494 @property
495 def _config_wrapper(self) -> ConfigWrapper:
496 return self._config_wrapper_stack.tail
498 @property
499 def _types_namespace(self) -> NamespacesTuple:
500 return self._ns_resolver.types_namespace
502 @property
503 def _arbitrary_types(self) -> bool:
504 return self._config_wrapper.arbitrary_types_allowed
506 # the following methods can be overridden but should be considered
507 # unstable / private APIs
508 def _list_schema(self, items_type: Any) -> CoreSchema:
509 return core_schema.list_schema(self.generate_schema(items_type))
511 def _dict_schema(self, keys_type: Any, values_type: Any) -> CoreSchema:
512 return core_schema.dict_schema(self.generate_schema(keys_type), self.generate_schema(values_type))
514 def _frozendict_schema(self, keys_type: Any, values_type: Any) -> CoreSchema:
515 return core_schema.frozendict_schema(self.generate_schema(keys_type), self.generate_schema(values_type))
517 def _ordered_dict_schema(self, keys_type: Any, values_type: Any) -> CoreSchema:
518 return core_schema.ordered_dict_schema(self.generate_schema(keys_type), self.generate_schema(values_type))
520 def _counter_schema(self, keys_type: Any) -> CoreSchema:
521 return core_schema.counter_schema(self.generate_schema(keys_type), core_schema.int_schema())
523 def _set_schema(self, items_type: Any) -> CoreSchema:
524 return core_schema.set_schema(self.generate_schema(items_type))
526 def _frozenset_schema(self, items_type: Any) -> CoreSchema:
527 return core_schema.frozenset_schema(self.generate_schema(items_type))
529 def _enum_schema(self, enum_type: type[Enum]) -> CoreSchema:
530 cases: list[Any] = list(enum_type.__members__.values())
532 enum_ref = _type_refs.enum_type_ref(enum_type)
533 description = None if not enum_type.__doc__ else inspect.cleandoc(enum_type.__doc__)
534 if (
535 description == 'An enumeration.'
536 ): # This is the default value provided by enum.EnumMeta.__new__; don't use it
537 description = None
538 js_updates = {'title': enum_type.__name__, 'description': description}
539 js_updates = {k: v for k, v in js_updates.items() if v is not None}
541 sub_type: Literal['str', 'int', 'float'] | None = None
542 if issubclass(enum_type, int):
543 sub_type = 'int'
544 value_ser_type: core_schema.SerSchema = core_schema.simple_ser_schema('int')
545 elif issubclass(enum_type, str):
546 # this handles `StrEnum` (3.11 only), and also `Foobar(str, Enum)`
547 sub_type = 'str'
548 value_ser_type = core_schema.simple_ser_schema('str')
549 elif issubclass(enum_type, float):
550 sub_type = 'float'
551 value_ser_type = core_schema.simple_ser_schema('float')
552 else:
553 # TODO this is an ugly hack, how do we trigger an Any schema for serialization?
554 value_ser_type = core_schema.plain_serializer_function_ser_schema(lambda x: x)
556 if cases:
558 def get_json_schema(schema: CoreSchema, handler: GetJsonSchemaHandler) -> JsonSchemaValue:
559 json_schema = handler(schema)
560 original_schema = handler.resolve_ref_schema(json_schema)
561 original_schema.update(js_updates)
562 return json_schema
564 enum_schema = core_schema.enum_schema(
565 enum_type,
566 cases,
567 sub_type=sub_type,
568 ref=enum_ref,
569 metadata={'pydantic_js_functions': [get_json_schema]},
570 )
572 if self._config_wrapper.use_enum_values:
573 enum_schema = core_schema.no_info_after_validator_function(
574 attrgetter('value'), enum_schema, serialization=value_ser_type
575 )
577 return enum_schema
579 else:
581 def get_json_schema_no_cases(_, handler: GetJsonSchemaHandler) -> JsonSchemaValue:
582 json_schema = handler(core_schema.enum_schema(enum_type, cases, sub_type=sub_type, ref=enum_ref))
583 original_schema = handler.resolve_ref_schema(json_schema)
584 original_schema.update(js_updates)
585 return json_schema
587 # Use an isinstance check for enums with no cases.
588 # The most important use case for this is creating TypeVar bounds for generics that should
589 # be restricted to enums. This is more consistent than it might seem at first, since you can only
590 # subclass enum.Enum (or subclasses of enum.Enum) if all parent classes have no cases.
591 # We use the get_json_schema function when an Enum subclass has been declared with no cases
592 # so that we can still generate a valid json schema.
593 return core_schema.is_instance_schema(
594 enum_type,
595 metadata={'pydantic_js_functions': [get_json_schema_no_cases]},
596 )
598 def _ip_schema(self, tp: Any) -> CoreSchema:
599 from ._validators import IP_VALIDATOR_LOOKUP, IpType
601 ip_type_json_schema_format: dict[type[IpType], str] = {
602 IPv4Address: 'ipv4',
603 IPv4Network: 'ipv4network',
604 IPv4Interface: 'ipv4interface',
605 IPv6Address: 'ipv6',
606 IPv6Network: 'ipv6network',
607 IPv6Interface: 'ipv6interface',
608 }
610 def ser_ip(ip: Any, info: core_schema.SerializationInfo) -> str | IpType:
611 if not isinstance(ip, (tp, str)):
612 raise PydanticSerializationUnexpectedValue(
613 f"Expected `{tp}` but got `{type(ip)}` with value `'{ip}'` - serialized value may not be as expected."
614 )
615 if info.mode == 'python':
616 return ip
617 return str(ip)
619 return core_schema.lax_or_strict_schema(
620 lax_schema=core_schema.no_info_plain_validator_function(IP_VALIDATOR_LOOKUP[tp]),
621 strict_schema=core_schema.json_or_python_schema(
622 json_schema=core_schema.no_info_after_validator_function(tp, core_schema.str_schema()),
623 python_schema=core_schema.is_instance_schema(tp),
624 ),
625 serialization=core_schema.plain_serializer_function_ser_schema(ser_ip, info_arg=True, when_used='always'),
626 metadata={
627 'pydantic_js_functions': [lambda _1, _2: {'type': 'string', 'format': ip_type_json_schema_format[tp]}]
628 },
629 )
631 def _path_schema(self, tp: Any, path_type: Any) -> CoreSchema:
632 if tp is os.PathLike and (path_type not in {str, bytes} and not typing_objects.is_any(path_type)):
633 raise PydanticUserError(
634 '`os.PathLike` can only be used with `str`, `bytes` or `Any`', code='schema-for-unknown-type'
635 )
637 path_constructor = pathlib.PurePath if tp is os.PathLike else tp
638 strict_inner_schema = (
639 core_schema.bytes_schema(strict=True) if (path_type is bytes) else core_schema.str_schema(strict=True)
640 )
641 lax_inner_schema = core_schema.bytes_schema() if (path_type is bytes) else core_schema.str_schema()
643 def path_validator(input_value: str | bytes) -> os.PathLike[Any]: # type: ignore
644 try:
645 if path_type is bytes:
646 if isinstance(input_value, bytes):
647 try:
648 input_value = input_value.decode()
649 except UnicodeDecodeError as e:
650 raise PydanticCustomError('bytes_type', 'Input must be valid bytes') from e
651 else:
652 raise PydanticCustomError('bytes_type', 'Input must be bytes')
653 elif not isinstance(input_value, str):
654 raise PydanticCustomError('path_type', 'Input is not a valid path')
656 return path_constructor(input_value) # type: ignore
657 except TypeError as e:
658 raise PydanticCustomError('path_type', 'Input is not a valid path') from e
660 def ser_path(path: Any, info: core_schema.SerializationInfo) -> str | os.PathLike[Any]:
661 if not isinstance(path, (tp, str)):
662 raise PydanticSerializationUnexpectedValue(
663 f"Expected `{tp}` but got `{type(path)}` with value `'{path}'` - serialized value may not be as expected."
664 )
665 if info.mode == 'python':
666 return path
667 return str(path)
669 instance_schema = core_schema.json_or_python_schema(
670 json_schema=core_schema.no_info_after_validator_function(path_validator, lax_inner_schema),
671 python_schema=core_schema.is_instance_schema(tp),
672 )
674 schema = core_schema.lax_or_strict_schema(
675 lax_schema=core_schema.union_schema(
676 [
677 instance_schema,
678 core_schema.no_info_after_validator_function(path_validator, strict_inner_schema),
679 ],
680 custom_error_type='path_type',
681 custom_error_message=f'Input is not a valid path for {tp}',
682 ),
683 strict_schema=instance_schema,
684 serialization=core_schema.plain_serializer_function_ser_schema(ser_path, info_arg=True, when_used='always'),
685 metadata={'pydantic_js_functions': [lambda source, handler: {**handler(source), 'format': 'path'}]},
686 )
687 return schema
689 def _deque_schema(self, items_type: Any) -> CoreSchema:
690 return core_schema.deque_schema(self.generate_schema(items_type))
692 def _mapping_schema(self, tp: Any, keys_type: Any, values_type: Any) -> CoreSchema:
693 from ._validators import MAPPING_ORIGIN_MAP, defaultdict_validator, get_defaultdict_default_default_factory
695 mapped_origin = MAPPING_ORIGIN_MAP[tp]
696 keys_schema = self.generate_schema(keys_type)
697 with warnings.catch_warnings():
698 # We kind of abused `Field()` default factories to be able to specify
699 # the `defaultdict`'s `default_factory`. As a consequence, we get warnings
700 # as normally `FieldInfo.default_factory` is unsupported in the context where
701 # `Field()` is used and our only solution is to ignore them (note that this might
702 # wrongfully ignore valid warnings, e.g. if the `value_type` is a PEP 695 type alias
703 # with unsupported metadata).
704 warnings.simplefilter('ignore', category=UnsupportedFieldAttributeWarning)
705 values_schema = self.generate_schema(values_type)
706 dict_schema = core_schema.dict_schema(keys_schema, values_schema, strict=False)
708 if mapped_origin is dict:
709 schema = dict_schema
710 else:
711 check_instance = core_schema.json_or_python_schema(
712 json_schema=dict_schema,
713 python_schema=core_schema.is_instance_schema(mapped_origin),
714 )
716 if tp is collections.defaultdict:
717 default_default_factory = get_defaultdict_default_default_factory(values_type)
718 coerce_instance_wrap = partial(
719 core_schema.no_info_wrap_validator_function,
720 partial(defaultdict_validator, default_default_factory=default_default_factory),
721 )
722 else:
723 coerce_instance_wrap = partial(core_schema.no_info_after_validator_function, mapped_origin)
725 lax_schema = coerce_instance_wrap(dict_schema)
726 strict_schema = core_schema.chain_schema([check_instance, lax_schema])
728 schema = core_schema.lax_or_strict_schema(
729 lax_schema=lax_schema,
730 strict_schema=strict_schema,
731 serialization=core_schema.wrap_serializer_function_ser_schema(
732 lambda v, h: h(v), schema=dict_schema, info_arg=False
733 ),
734 )
736 return schema
738 def _arbitrary_type_schema(self, tp: Any) -> CoreSchema:
739 if not isinstance(tp, type):
740 warnings.warn(
741 f'{tp!r} is not a Python type (it may be an instance of an object),'
742 ' Pydantic will allow any object with no validation since we cannot even'
743 ' enforce that the input is an instance of the given type.'
744 ' To get rid of this error wrap the type with `pydantic.SkipValidation`.',
745 ArbitraryTypeWarning,
746 )
747 return core_schema.any_schema()
748 return core_schema.is_instance_schema(tp)
750 def _unknown_type_schema(self, obj: Any) -> CoreSchema:
751 raise PydanticSchemaGenerationError(
752 f'Unable to generate pydantic-core schema for {obj!r}. '
753 'Set `arbitrary_types_allowed=True` in the model_config to ignore this error'
754 ' or implement `__get_pydantic_core_schema__` on your type to fully support it.'
755 '\n\nIf you got this error by calling handler(<some type>) within'
756 ' `__get_pydantic_core_schema__` then you likely need to call'
757 ' `handler.generate_schema(<some type>)` since we do not call'
758 ' `__get_pydantic_core_schema__` on `<some type>` otherwise to avoid infinite recursion.'
759 )
761 def _apply_discriminator_to_union(
762 self, schema: CoreSchema, discriminator: str | Discriminator | None
763 ) -> CoreSchema:
764 if discriminator is None:
765 return schema
766 try:
767 return _discriminated_union.apply_discriminator(
768 schema,
769 discriminator,
770 self.defs._definitions,
771 )
772 except _discriminated_union.MissingDefinitionForUnionRef:
773 # defer until defs are resolved
774 _discriminated_union.set_discriminator_in_metadata(
775 schema,
776 discriminator,
777 )
778 return schema
780 def clean_schema(self, schema: CoreSchema) -> CoreSchema:
781 return self.defs.finalize_schema(schema)
783 def _add_js_function(self, metadata_schema: CoreSchema, js_function: Callable[..., Any]) -> None:
784 metadata = metadata_schema.get('metadata', {})
785 pydantic_js_functions = metadata.setdefault('pydantic_js_functions', [])
786 # because of how we generate core schemas for nested generic models
787 # we can end up adding `BaseModel.__get_pydantic_json_schema__` multiple times
788 # this check may fail to catch duplicates if the function is a `functools.partial`
789 # or something like that, but if it does it'll fail by inserting the duplicate
790 if js_function not in pydantic_js_functions:
791 pydantic_js_functions.append(js_function)
792 metadata_schema['metadata'] = metadata
794 def generate_schema(
795 self,
796 obj: Any,
797 ) -> core_schema.CoreSchema:
798 """Generate core schema.
800 Args:
801 obj: The object to generate core schema for.
803 Returns:
804 The generated core schema.
806 Raises:
807 PydanticUndefinedAnnotation:
808 If it is not possible to evaluate forward reference.
809 PydanticSchemaGenerationError:
810 If it is not possible to generate pydantic-core schema.
811 TypeError:
812 - If `alias_generator` returns a disallowed type (must be str, AliasPath or AliasChoices).
813 - If V1 style validator with `each_item=True` applied on a wrong field.
814 PydanticUserError:
815 - If `typing.TypedDict` is used instead of `typing_extensions.TypedDict` on Python < 3.12.
816 - If `__modify_schema__` method is used instead of `__get_pydantic_json_schema__`.
817 """
818 origin = get_origin(obj)
819 schema: CoreSchema | None = None
820 if origin is None:
821 # If the type has an origin, it is guaranteed to *not* be an ordinary class that could have
822 # a `__get_pydantic_core_schema__()` method:
823 schema = self._generate_schema_from_get_schema_method(obj, obj)
825 if schema is None:
826 schema = self._generate_schema_inner(obj, origin)
828 # `__get_pydantic_json_schema__()` is checked on the origin if it exists:
829 metadata_js_function = _extract_get_pydantic_json_schema(origin if origin is not None else obj)
830 if metadata_js_function is not None:
831 metadata_schema = resolve_original_schema(schema, self.defs)
832 if metadata_schema:
833 self._add_js_function(metadata_schema, metadata_js_function)
835 schema = _add_custom_serialization_from_json_encoders(self._config_wrapper.json_encoders, obj, schema)
837 return schema
839 def _model_schema(self, cls: type[BaseModel]) -> core_schema.CoreSchema:
840 """Generate schema for a Pydantic model."""
841 BaseModel_ = import_cached_base_model()
843 model_ref = _type_refs.model_type_ref(cls)
844 with self.defs.get_definition_ref_schema(ref=model_ref) as maybe_schema:
845 if maybe_schema is not None:
846 return maybe_schema
848 schema = cls.__dict__.get('__pydantic_core_schema__')
849 if schema is not None and not isinstance(schema, MockCoreSchema):
850 if schema['type'] == 'definitions':
851 schema = self.defs.unpack_definitions(schema)
852 ref = get_ref(schema)
853 if ref:
854 return self.defs.create_definition_reference_schema(schema)
855 else:
856 return schema
858 config_wrapper = ConfigWrapper(cls.model_config, check=False)
860 with self._config_wrapper_stack.push(config_wrapper), self._ns_resolver.push(cls):
861 core_config = self._config_wrapper.core_config(title=cls.__name__)
863 if cls.__pydantic_fields_complete__ or cls is BaseModel_:
864 fields = getattr(cls, '__pydantic_fields__', {})
865 extra_info = getattr(cls, '__pydantic_extra_info__', None)
866 else:
867 if '__pydantic_fields__' not in cls.__dict__:
868 # This happens when we have a loop in the schema generation:
869 # class Base[T](BaseModel):
870 # t: T
871 #
872 # class Other(BaseModel):
873 # b: 'Base[Other]'
874 # When we build fields for `Other`, we evaluate the forward annotation.
875 # At this point, `Other` doesn't have the model fields set. We create
876 # `Base[Other]`; model fields are successfully built, and we try to generate
877 # a schema for `t: Other`. As `Other.__pydantic_fields__` aren't set, we abort.
878 raise PydanticUndefinedAnnotation(
879 name=cls.__name__,
880 message=f'Class {cls.__name__!r} is not defined',
881 )
882 try:
883 fields, extra_info = rebuild_model_fields(
884 cls,
885 config_wrapper=self._config_wrapper,
886 ns_resolver=self._ns_resolver,
887 typevars_map=self._typevars_map or {},
888 )
889 except NameError as e:
890 raise PydanticUndefinedAnnotation.from_name_error(e) from e
892 decorators = cls.__pydantic_decorators__
893 computed_fields = decorators.computed_fields
894 check_decorator_fields_exist(
895 chain(
896 decorators.field_validators.values(),
897 decorators.field_serializers.values(),
898 decorators.validators.values(),
899 ),
900 {*fields.keys(), *computed_fields.keys()},
901 )
903 model_validators = decorators.model_validators.values()
905 extras_schema = None
906 extras_keys_schema = None
907 if core_config.get('extra_fields_behavior') == 'allow' and extra_info is not None:
908 tp = get_origin(extra_info.annotation)
909 if tp is not dict:
910 raise PydanticSchemaGenerationError(
911 'The type annotation for `__pydantic_extra__` must be `dict[str, ...]`'
912 )
913 # See the comments in `_get_args_resolving_forward_refs()` for why we need
914 # to re-evaluate the annotation:
915 extra_keys_type, extra_items_type = self._get_args_resolving_forward_refs(
916 extra_info.annotation,
917 required=True,
918 )
919 if extra_keys_type is not str:
920 extras_keys_schema = self.generate_schema(extra_keys_type)
921 if not typing_objects.is_any(extra_items_type):
922 extras_schema = self.generate_schema(extra_items_type)
924 generic_origin: type[BaseModel] | None = getattr(cls, '__pydantic_generic_metadata__', {}).get('origin')
926 if cls.__pydantic_root_model__:
927 inner_schema, metadata = self._common_field_schema('root', fields['root'], decorators)
928 if cls.__doc__ and metadata.get('pydantic_js_updates', {}).get('description'):
929 # This is a bit of a leaky abstraction, but as the model docstring takes priority
930 # over the root field's description, we need to override it here. This can't be done
931 # in the JSON Schema generation logic because the metadata's `pydantic_js_updates` are
932 # applied last, and overrides any value previously set (so the description set from the
933 # docstring in `GenerateJsonSchema._update_class_schema()` is overridden):
934 update_core_metadata(
935 metadata, pydantic_js_updates={'description': inspect.cleandoc(cls.__doc__)}
936 )
938 inner_schema = apply_model_validators(inner_schema, model_validators, 'inner')
939 model_schema = core_schema.model_schema(
940 cls,
941 inner_schema,
942 generic_origin=generic_origin,
943 custom_init=getattr(cls, '__pydantic_custom_init__', None),
944 root_model=True,
945 post_init=getattr(cls, '__pydantic_post_init__', None),
946 config=core_config,
947 ref=model_ref,
948 metadata=metadata,
949 )
950 else:
951 fields_schema: core_schema.CoreSchema = core_schema.model_fields_schema(
952 {k: self._generate_md_field_schema(k, v, decorators) for k, v in fields.items()},
953 computed_fields=[
954 self._computed_field_schema(d, decorators.field_serializers)
955 for d in computed_fields.values()
956 ],
957 extras_schema=extras_schema,
958 extras_keys_schema=extras_keys_schema,
959 model_name=cls.__name__,
960 )
961 inner_schema = apply_validators(fields_schema, decorators.root_validators.values())
962 inner_schema = apply_model_validators(inner_schema, model_validators, 'inner')
964 model_schema = core_schema.model_schema(
965 cls,
966 inner_schema,
967 generic_origin=generic_origin,
968 custom_init=getattr(cls, '__pydantic_custom_init__', None),
969 root_model=False,
970 post_init=getattr(cls, '__pydantic_post_init__', None),
971 config=core_config,
972 ref=model_ref,
973 )
975 schema = self._apply_model_serializers(model_schema, decorators.model_serializers.values())
976 schema = apply_model_validators(schema, model_validators, 'outer')
977 return self.defs.create_definition_reference_schema(schema)
979 def _resolve_self_type(self, obj: Any) -> Any:
980 obj = self.model_type_stack.get()
981 if obj is None:
982 raise PydanticUserError('`typing.Self` is invalid in this context', code='invalid-self-type')
983 return obj
985 def _generate_schema_from_get_schema_method(self, obj: Any, source: Any) -> core_schema.CoreSchema | None:
986 BaseModel_ = import_cached_base_model()
988 get_schema = getattr(obj, '__get_pydantic_core_schema__', None)
990 if (
991 get_schema is not None
992 # BaseModel.__get_pydantic_core_schema__ is defined for backwards compatibility,
993 # to allow existing code to call `super().__get_pydantic_core_schema__` in Pydantic
994 # model that overrides `__get_pydantic_core_schema__`. However, it raises a deprecation
995 # warning stating that the method will be removed, and during the core schema gen we actually
996 # don't call the method:
997 and getattr(get_schema, '__func__', None) is not BaseModel_.__get_pydantic_core_schema__.__func__ # pyright: ignore[reportFunctionMemberAccess]
998 ):
999 # Some referenceable types might have a `__get_pydantic_core_schema__` method
1000 # defined on it by users (e.g. on a dataclass). This generally doesn't play well
1001 # as these types are already recognized by the `GenerateSchema` class and isn't ideal
1002 # as we might end up calling `any_type_ref()` (relatively expensive) on types that are actually
1003 # not referenceable:
1004 with self.defs.get_definition_ref_schema(ref=_type_refs.any_type_ref(obj, get_origin(obj))) as maybe_schema:
1005 if maybe_schema is not None:
1006 return maybe_schema
1008 if obj is source:
1009 ref_mode = 'unpack'
1010 else:
1011 ref_mode = 'to-def'
1012 schema = get_schema(
1013 source, CallbackGetCoreSchemaHandler(self._generate_schema_inner, self, ref_mode=ref_mode)
1014 )
1015 if schema['type'] == 'definitions':
1016 schema = self.defs.unpack_definitions(schema)
1018 ref = get_ref(schema)
1019 if ref:
1020 return self.defs.create_definition_reference_schema(schema)
1022 # Note: if schema is of type `'definition-ref'`, we might want to copy it as a
1023 # safety measure (because these are inlined in place -- i.e. mutated directly)
1024 return schema
1026 if get_schema is None and (validators := getattr(obj, '__get_validators__', None)) is not None:
1027 from pydantic.v1 import BaseModel as BaseModelV1
1029 if issubclass(obj, BaseModelV1):
1030 warnings.warn(
1031 f'Mixing V1 models and V2 models (or constructs, like `TypeAdapter`) is not supported. Please upgrade `{obj.__name__}` to V2.',
1032 UserWarning,
1033 )
1034 else:
1035 warnings.warn(
1036 '`__get_validators__` is deprecated and will be removed, use `__get_pydantic_core_schema__` instead.',
1037 PydanticDeprecatedSince20,
1038 )
1039 return core_schema.chain_schema([core_schema.with_info_plain_validator_function(v) for v in validators()])
1041 def _resolve_forward_ref(self, obj: Any) -> Any:
1042 # we assume that types_namespace has the target of forward references in its scope,
1043 # but this could fail, for example, if calling Validator on an imported type which contains
1044 # forward references to other types only defined in the module from which it was imported
1045 # `Validator(SomeImportedTypeAliasWithAForwardReference)`
1046 # or the equivalent for BaseModel
1047 # class Model(BaseModel):
1048 # x: SomeImportedTypeAliasWithAForwardReference
1049 try:
1050 obj = _typing_extra.eval_type(obj, *self._types_namespace)
1051 except NameError as e:
1052 raise PydanticUndefinedAnnotation.from_name_error(e) from e
1054 # if obj is still a ForwardRef, it means we can't evaluate it, raise PydanticUndefinedAnnotation
1055 if isinstance(obj, ForwardRef):
1056 raise PydanticUndefinedAnnotation(obj.__forward_arg__, f'Unable to evaluate forward reference {obj}')
1058 if self._typevars_map:
1059 obj = replace_types(obj, self._typevars_map)
1061 return obj
1063 @overload
1064 def _get_args_resolving_forward_refs(self, obj: Any, required: Literal[True]) -> tuple[Any, ...]: ...
1066 @overload
1067 def _get_args_resolving_forward_refs(self, obj: Any) -> tuple[Any, ...] | None: ...
1069 def _get_args_resolving_forward_refs(self, obj: Any, required: bool = False) -> tuple[Any, ...] | None:
1070 args = get_args(obj)
1071 if args:
1072 if isinstance(obj, GenericAlias):
1073 # PEP 585 generic aliases don't convert args to ForwardRefs, unlike `typing.List/Dict` etc.
1074 # This was fixed in https://github.com/python/cpython/pull/30900 (Python 3.11).
1075 # TODO: this shouldn't be necessary (probably even this `_get_args_resolving_forward_refs()` function)
1076 # once we drop support for Python 3.10 *or* if we implement our own `typing._eval_type()` implementation.
1077 args = (ForwardRef(a) if isinstance(a, str) else a for a in args)
1078 args = tuple(self._resolve_forward_ref(a) if isinstance(a, ForwardRef) else a for a in args)
1079 elif required: # pragma: no cover
1080 raise TypeError(f'Expected {obj} to have generic parameters but it had none')
1081 return args
1083 def _get_first_arg_or_any(self, obj: Any) -> Any:
1084 args = self._get_args_resolving_forward_refs(obj)
1085 if not args:
1086 return Any
1087 return args[0]
1089 def _get_first_two_args_or_any(self, obj: Any) -> tuple[Any, Any]:
1090 args = self._get_args_resolving_forward_refs(obj)
1091 if not args:
1092 return (Any, Any)
1093 if len(args) < 2:
1094 origin = get_origin(obj)
1095 raise TypeError(f'Expected two type arguments for {origin}, got 1')
1096 return args[0], args[1]
1098 def _generate_schema_inner(self, obj: Any, origin: Any | None = PydanticUndefined) -> core_schema.CoreSchema:
1099 if typing_objects.is_self(obj):
1100 obj = self._resolve_self_type(obj)
1102 origin = origin if origin is not PydanticUndefined else get_origin(obj)
1104 if is_union_origin(origin):
1105 return self._union_schema(obj)
1107 if typing_objects.is_annotated(origin):
1108 return self._annotated_schema(obj)
1110 if typing_objects.is_literal(origin):
1111 return self._literal_schema(obj)
1113 if isinstance(obj, dict):
1114 # we assume this is already a valid schema
1115 return obj # type: ignore[return-value]
1117 if isinstance(obj, str):
1118 obj = ForwardRef(obj)
1120 if isinstance(obj, ForwardRef):
1121 return self.generate_schema(self._resolve_forward_ref(obj))
1123 BaseModel = import_cached_base_model()
1125 if lenient_issubclass(obj, BaseModel):
1126 with self.model_type_stack.push(obj):
1127 return self._model_schema(obj)
1129 if isinstance(obj, PydanticRecursiveRef):
1130 return core_schema.definition_reference_schema(schema_ref=obj.type_ref)
1132 return self.match_type(obj, origin)
1134 def match_type(self, obj: Any, origin: Any) -> core_schema.CoreSchema: # noqa: C901
1135 """Main mapping of types to schemas.
1137 The general structure is a series of if statements starting with the simple cases
1138 (non-generic primitive types) and then handling generics and other more complex cases.
1140 Each case either generates a schema directly, calls into a public user-overridable method
1141 (like `GenerateSchema.tuple_variable_schema`) or calls into a private method that handles some
1142 boilerplate before calling into the user-facing method (e.g. `GenerateSchema._tuple_schema`).
1144 The idea is that we'll evolve this into adding more and more user facing methods over time
1145 as they get requested and we figure out what the right API for them is.
1146 """
1147 if typing_objects.is_any(obj) or typing_objects.is_final(obj):
1148 # TODO: I'm not sure obj can be `Final` here, we call `typing_inspection.introspection.inspect_annotation()`
1149 # early for model-like fields, which handles it already.
1150 return core_schema.any_schema()
1152 if origin is not None and (aliased_obj := typing_objects.DEPRECATED_ALIASES_IDS.get(id(obj))):
1153 # See https://typing-inspection.pydantic.dev/dev/usage/#inspecting-the-type-expression:
1154 obj = aliased_obj
1155 origin = None
1157 if origin is None:
1158 # Thanks to the checks before this point, no origin means `obj` is a
1159 # simple type (not a generic alias, a union, an `Annotated` form).
1160 # examples: int, str, list, dict, collections.abc.Mapping, a `TypeAliasType` instance.
1161 try:
1162 handler = self._handlers.get(obj)
1163 except TypeError:
1164 pass
1165 else:
1166 if handler is not None:
1167 return handler(self, obj)
1169 if typing_objects.is_typealiastype(obj):
1170 return self._type_alias_type_schema(obj)
1171 elif is_typeddict(obj):
1172 return self._typed_dict_schema(obj, None)
1173 elif inspect.isclass(obj) and issubclass(obj, Enum):
1174 # NOTE: this must come before the `is_namedtuple()` check as enums values
1175 # can be namedtuples:
1176 return self._enum_schema(obj)
1177 elif _typing_extra.is_namedtuple(obj):
1178 return self._namedtuple_schema(obj, None)
1179 elif typing_objects.is_newtype(obj):
1180 return self.generate_schema(obj.__supertype__)
1181 elif isinstance(obj, typing.TypeVar):
1182 return self._unsubstituted_typevar_schema(obj)
1183 elif isinstance(obj, VALIDATE_CALL_SUPPORTED_TYPES):
1184 return self._call_schema(obj) # pyright: ignore[reportArgumentType]
1185 # dataclasses.is_dataclass coerces dc instances to types, but we only handle
1186 # the case of a dc type here
1187 elif dataclasses.is_dataclass(obj):
1188 return self._dataclass_schema(obj, None) # pyright: ignore[reportArgumentType]
1190 if origin is not None:
1191 return self._match_generic_type(obj, origin)
1193 if self._arbitrary_types:
1194 return self._arbitrary_type_schema(obj)
1195 return self._unknown_type_schema(obj)
1197 def _match_generic_type(self, obj: Any, origin: Any) -> CoreSchema: # noqa: C901
1198 # Need to handle generic dataclasses before looking for the schema properties because attribute accesses
1199 # on _GenericAlias delegate to the origin type, so lose the information about the concrete parametrization
1200 # As a result, currently, there is no way to cache the schema for generic dataclasses. This may be possible
1201 # to resolve by modifying the value returned by `Generic.__class_getitem__`, but that is a dangerous game.
1202 if dataclasses.is_dataclass(origin):
1203 return self._dataclass_schema(obj, origin) # pyright: ignore[reportArgumentType]
1204 elif _typing_extra.is_namedtuple(origin):
1205 return self._namedtuple_schema(obj, origin)
1206 elif typing_objects.is_final(origin):
1207 # TODO: I'm not sure obj can be `Final` here, we call `typing_inspection.introspection.inspect_annotation()`
1208 # early for model-like fields, which handles it already.
1209 return self.generate_schema(self._get_first_arg_or_any(obj))
1211 schema = self._generate_schema_from_get_schema_method(origin, obj)
1212 if schema is not None:
1213 return schema
1215 try:
1216 handler = self._generic_handlers.get(origin)
1217 except TypeError:
1218 pass
1219 else:
1220 if handler is not None:
1221 return handler(self, obj)
1223 if typing_objects.is_typealiastype(origin):
1224 return self._type_alias_type_schema(obj)
1225 elif is_typeddict(origin):
1226 return self._typed_dict_schema(obj, origin)
1228 if self._arbitrary_types:
1229 return self._arbitrary_type_schema(origin)
1230 return self._unknown_type_schema(obj)
1232 def _generate_td_field_schema(
1233 self,
1234 name: str,
1235 field_info: FieldInfo,
1236 decorators: DecoratorInfos,
1237 *,
1238 required: bool = True,
1239 ) -> core_schema.TypedDictField:
1240 """Prepare a TypedDictField to represent a model or typeddict field."""
1241 schema, metadata = self._common_field_schema(name, field_info, decorators)
1242 return core_schema.typed_dict_field(
1243 schema,
1244 required=False if not field_info.is_required() else required,
1245 serialization_exclude=field_info.exclude,
1246 validation_alias=_convert_to_aliases(field_info.validation_alias),
1247 serialization_alias=field_info.serialization_alias,
1248 serialization_exclude_if=field_info.exclude_if,
1249 metadata=metadata,
1250 )
1252 def _generate_md_field_schema(
1253 self,
1254 name: str,
1255 field_info: FieldInfo,
1256 decorators: DecoratorInfos,
1257 ) -> core_schema.ModelField:
1258 """Prepare a ModelField to represent a model field."""
1259 schema, metadata = self._common_field_schema(name, field_info, decorators)
1260 return core_schema.model_field(
1261 schema,
1262 serialization_exclude=field_info.exclude,
1263 validation_alias=_convert_to_aliases(field_info.validation_alias),
1264 serialization_alias=field_info.serialization_alias,
1265 serialization_exclude_if=field_info.exclude_if,
1266 frozen=field_info.frozen,
1267 metadata=metadata,
1268 )
1270 def _generate_dc_field_schema(
1271 self,
1272 name: str,
1273 field_info: FieldInfo,
1274 decorators: DecoratorInfos,
1275 ) -> core_schema.DataclassField:
1276 """Prepare a DataclassField to represent the parameter/field, of a dataclass."""
1277 schema, metadata = self._common_field_schema(name, field_info, decorators)
1278 return core_schema.dataclass_field(
1279 name,
1280 schema,
1281 init=field_info.init,
1282 init_only=field_info.init_var or None,
1283 kw_only=None if field_info.kw_only else False,
1284 serialization_exclude=field_info.exclude,
1285 validation_alias=_convert_to_aliases(field_info.validation_alias),
1286 serialization_alias=field_info.serialization_alias,
1287 serialization_exclude_if=field_info.exclude_if,
1288 frozen=field_info.frozen,
1289 metadata=metadata,
1290 )
1292 def _common_field_schema( # C901
1293 self, name: str, field_info: FieldInfo, decorators: DecoratorInfos
1294 ) -> tuple[CoreSchema, dict[str, Any]]:
1295 source_type, annotations = field_info.annotation, field_info.metadata
1297 def set_discriminator(schema: CoreSchema) -> CoreSchema:
1298 schema = self._apply_discriminator_to_union(schema, field_info.discriminator)
1299 return schema
1301 # Convert `@field_validator` decorators to `Before/After/Plain/WrapValidator` instances:
1302 if decorators.field_validators:
1303 validators_from_decorators = [
1304 _mode_to_validator[decorator.info.mode]._from_decorator(decorator)
1305 for decorator in filter_field_decorator_info_by_field(decorators.field_validators.values(), name)
1306 ]
1307 else:
1308 validators_from_decorators = []
1310 with self.field_name_stack.push(name):
1311 if field_info.discriminator is not None:
1312 schema = self._apply_annotations(
1313 source_type, annotations + validators_from_decorators, transform_inner_schema=set_discriminator
1314 )
1315 else:
1316 schema = self._apply_annotations(
1317 source_type,
1318 annotations + validators_from_decorators,
1319 )
1321 if decorators.validators:
1322 # This V1 compatibility shim should eventually be removed
1323 # push down any `each_item=True` validators
1324 # note that this won't work for any Annotated types that get wrapped by a function validator
1325 # but that's okay because that didn't exist in V1
1326 this_field_validators = filter_field_decorator_info_by_field(decorators.validators.values(), name)
1327 if _validators_require_validate_default(this_field_validators):
1328 field_info.validate_default = True
1329 each_item_validators = [v for v in this_field_validators if v.info.each_item is True]
1330 this_field_validators = [v for v in this_field_validators if v not in each_item_validators]
1331 schema = apply_each_item_validators(schema, each_item_validators)
1333 schema = apply_validators(schema, this_field_validators)
1335 # the default validator needs to go outside of any other validators
1336 # so that it is the topmost validator for the field validator
1337 # which uses it to check if the field has a default value or not
1338 if not field_info.is_required():
1339 schema = wrap_default(field_info, schema)
1341 if decorators.field_serializers:
1342 schema = self._apply_field_serializers(
1343 schema, filter_field_decorator_info_by_field(decorators.field_serializers.values(), name)
1344 )
1346 pydantic_js_updates, pydantic_js_extra = _extract_json_schema_info_from_field_info(field_info)
1347 core_metadata: dict[str, Any] = {}
1348 update_core_metadata(
1349 core_metadata, pydantic_js_updates=pydantic_js_updates, pydantic_js_extra=pydantic_js_extra
1350 )
1352 return schema, core_metadata
1354 def _union_schema(self, union_type: Any) -> core_schema.CoreSchema:
1355 """Generate schema for a Union."""
1356 args = self._get_args_resolving_forward_refs(union_type, required=True)
1357 choices: list[CoreSchema] = []
1358 nullable = False
1359 allow_missing = False
1360 for arg in args:
1361 if arg is None or arg is NoneType:
1362 nullable = True
1363 elif arg is MISSING:
1364 allow_missing = True
1365 else:
1366 choices.append(self.generate_schema(arg))
1368 if not choices:
1369 # Only `None` and `MISSING` were present in the union (e.g. `None | MISSING`):
1370 s = core_schema.none_schema()
1371 nullable = False
1372 elif len(choices) == 1:
1373 s = choices[0]
1374 else:
1375 choices_with_tags: list[CoreSchema | tuple[CoreSchema, str]] = []
1376 for choice in choices:
1377 tag = cast(CoreMetadata, choice.get('metadata', {})).get('pydantic_internal_union_tag_key')
1378 if tag is not None:
1379 choices_with_tags.append((choice, tag))
1380 else:
1381 choices_with_tags.append(choice)
1382 s = core_schema.union_schema(choices_with_tags)
1384 if nullable:
1385 s = core_schema.nullable_schema(s)
1386 if allow_missing:
1387 s = core_schema.missing_sentinel_schema(s)
1388 return s
1390 def _type_alias_type_schema(self, obj: TypeAliasType) -> CoreSchema:
1391 ref = _type_refs.type_alias_type_ref(obj)
1392 with self.defs.get_definition_ref_schema(ref=ref) as maybe_schema:
1393 if maybe_schema is not None:
1394 return maybe_schema
1396 origin: TypeAliasType = get_origin(obj) or obj
1397 typevars_map = get_standard_typevars_map(obj)
1399 with self._ns_resolver.push(origin):
1400 try:
1401 annotation = _typing_extra.eval_type(origin.__value__, *self._types_namespace)
1402 except NameError as e:
1403 raise PydanticUndefinedAnnotation.from_name_error(e) from e
1404 annotation = replace_types(annotation, typevars_map)
1405 schema = self.generate_schema(annotation)
1406 assert schema['type'] != 'definitions'
1407 schema['ref'] = ref # type: ignore
1408 return self.defs.create_definition_reference_schema(schema)
1410 def _literal_schema(self, literal_type: Any) -> CoreSchema:
1411 """Generate schema for a Literal."""
1412 expected = list(get_literal_values(literal_type, type_check=False, unpack_type_aliases='eager'))
1413 assert expected, f'literal "expected" cannot be empty, obj={literal_type}'
1414 schema = core_schema.literal_schema(expected)
1416 if self._config_wrapper.use_enum_values and any(isinstance(v, Enum) for v in expected):
1417 schema = core_schema.no_info_after_validator_function(
1418 lambda v: v.value if isinstance(v, Enum) else v, schema
1419 )
1421 return schema
1423 def _typed_dict_schema(self, typed_dict_cls: Any, origin: Any) -> core_schema.CoreSchema:
1424 """Generate a core schema for a `TypedDict` class.
1426 To be able to build a `DecoratorInfos` instance for the `TypedDict` class (which will include
1427 validators, serializers, etc.), we need to have access to the original bases of the class
1428 (see https://docs.python.org/3/library/types.html#types.get_original_bases).
1429 However, the `__orig_bases__` attribute was only added in 3.12 (https://github.com/python/cpython/pull/103698).
1431 For this reason, we require Python 3.12 (or using the `typing_extensions` backport).
1432 """
1433 FieldInfo = import_cached_field_info()
1435 typed_dict_ref = _type_refs.class_type_ref(typed_dict_cls, origin)
1436 with (
1437 self.model_type_stack.push(typed_dict_cls),
1438 self.defs.get_definition_ref_schema(ref=typed_dict_ref) as maybe_schema,
1439 ):
1440 if maybe_schema is not None:
1441 return maybe_schema
1443 typevars_map = get_standard_typevars_map(typed_dict_cls)
1444 if origin is not None:
1445 typed_dict_cls = origin
1447 if not _SUPPORTS_TYPEDDICT and type(typed_dict_cls).__module__ == 'typing':
1448 raise PydanticUserError(
1449 'Please use `typing_extensions.TypedDict` instead of `typing.TypedDict` on Python < 3.12.',
1450 code='typed-dict-version',
1451 )
1453 try:
1454 # if a typed dictionary class doesn't have config, we use the parent's config, hence a default of `None`
1455 # see https://github.com/pydantic/pydantic/issues/10917
1456 config: ConfigDict | None = get_attribute_from_bases(typed_dict_cls, '__pydantic_config__')
1457 except AttributeError:
1458 config = None
1460 with self._config_wrapper_stack.push(config):
1461 core_config = self._config_wrapper.core_config(title=typed_dict_cls.__name__)
1463 required_keys: frozenset[str] = typed_dict_cls.__required_keys__
1465 fields: dict[str, core_schema.TypedDictField] = {}
1467 decorators = DecoratorInfos.build(typed_dict_cls, replace_wrapped_methods=False)
1468 decorators.update_from_config(self._config_wrapper)
1470 if self._config_wrapper.use_attribute_docstrings:
1471 field_docstrings = extract_docstrings_from_cls(typed_dict_cls, use_inspect=True)
1472 else:
1473 field_docstrings = None
1475 try:
1476 annotations = _typing_extra.get_cls_type_hints(typed_dict_cls, ns_resolver=self._ns_resolver)
1477 except NameError as e:
1478 raise PydanticUndefinedAnnotation.from_name_error(e) from e
1480 readonly_fields: list[str] = []
1482 for field_name, annotation in annotations.items():
1483 field_info = FieldInfo.from_annotation(annotation, _source=AnnotationSource.TYPED_DICT)
1484 field_info.annotation = replace_types(field_info.annotation, typevars_map)
1486 required = (
1487 field_name in required_keys or 'required' in field_info._qualifiers
1488 ) and 'not_required' not in field_info._qualifiers
1489 if 'read_only' in field_info._qualifiers:
1490 readonly_fields.append(field_name)
1492 if (
1493 field_docstrings is not None
1494 and field_info.description is None
1495 and field_name in field_docstrings
1496 ):
1497 field_info.description = field_docstrings[field_name]
1498 update_field_from_config(self._config_wrapper, field_name, field_info)
1500 fields[field_name] = self._generate_td_field_schema(
1501 field_name, field_info, decorators, required=required
1502 )
1504 if readonly_fields:
1505 fields_repr = ', '.join(repr(f) for f in readonly_fields)
1506 plural = len(readonly_fields) >= 2
1507 warnings.warn(
1508 f'Item{"s" if plural else ""} {fields_repr} on TypedDict class {typed_dict_cls.__name__!r} '
1509 f'{"are" if plural else "is"} using the `ReadOnly` qualifier. Pydantic will not protect items '
1510 'from any mutation on dictionary instances.',
1511 UserWarning,
1512 )
1514 extra_behavior: core_schema.ExtraBehavior = 'ignore'
1515 extras_schema: CoreSchema | None = None # For 'allow', equivalent to `Any` - no validation performed.
1517 is_closed, extra_items, extra_items_readonly = _get_typed_dict_extra_items(typed_dict_cls)
1518 if is_closed:
1519 extra_behavior = 'forbid'
1520 extras_schema = None
1521 elif not typing_objects.is_noextraitems(extra_items):
1522 extra_behavior = 'allow'
1523 extras_schema = self.generate_schema(replace_types(extra_items, typevars_map))
1525 if extra_items_readonly:
1526 warnings.warn(
1527 f'Extra items on TypedDict class {typed_dict_cls.__name__!r} are using the `ReadOnly` qualifier. '
1528 'Pydantic will not protect items from any mutation on dictionary instances.',
1529 UserWarning,
1530 )
1532 if (config_extra := self._config_wrapper.extra) in ('allow', 'forbid'):
1533 if is_closed and config_extra == 'allow':
1534 warnings.warn(
1535 f"TypedDict class {typed_dict_cls.__qualname__!r} is closed, but 'extra' configuration "
1536 "is set to `'allow'`. The 'extra' configuration value will be ignored.",
1537 category=TypedDictExtraConfigWarning,
1538 )
1539 elif not typing_objects.is_noextraitems(extra_items) and config_extra == 'forbid':
1540 warnings.warn(
1541 f"TypedDict class {typed_dict_cls.__qualname__!r} allows extra items, but 'extra' configuration "
1542 "is set to `'forbid'`. The 'extra' configuration value will be ignored.",
1543 category=TypedDictExtraConfigWarning,
1544 )
1545 else:
1546 extra_behavior = config_extra
1548 td_schema = core_schema.typed_dict_schema(
1549 fields,
1550 cls=typed_dict_cls,
1551 computed_fields=[
1552 self._computed_field_schema(d, decorators.field_serializers)
1553 for d in decorators.computed_fields.values()
1554 ],
1555 extra_behavior=extra_behavior,
1556 extras_schema=extras_schema,
1557 ref=typed_dict_ref,
1558 config=core_config,
1559 )
1561 schema = self._apply_model_serializers(td_schema, decorators.model_serializers.values())
1562 schema = apply_model_validators(schema, decorators.model_validators.values(), 'all')
1563 return self.defs.create_definition_reference_schema(schema)
1565 def _namedtuple_schema(self, namedtuple_cls: Any, origin: Any) -> core_schema.CoreSchema:
1566 """Generate schema for a NamedTuple."""
1567 namedtuple_ref = _type_refs.class_type_ref(namedtuple_cls, origin)
1568 with (
1569 self.model_type_stack.push(namedtuple_cls),
1570 self.defs.get_definition_ref_schema(ref=namedtuple_ref) as maybe_schema,
1571 ):
1572 if maybe_schema is not None:
1573 return maybe_schema
1574 typevars_map = get_standard_typevars_map(namedtuple_cls)
1575 if origin is not None:
1576 namedtuple_cls = origin
1578 try:
1579 annotations = _typing_extra.get_cls_type_hints(namedtuple_cls, ns_resolver=self._ns_resolver)
1580 except NameError as e:
1581 raise PydanticUndefinedAnnotation.from_name_error(e) from e
1583 # Filter annotations to only include fields that are actually in the NamedTuple
1584 # (as subclassing an existing NamedTuple is not supported yet - see https://github.com/python/typing/issues/427)
1585 # and use `Any` if no annotation exist (i.e. when using `collections.namedtuple()`).
1586 annotations = {field_name: annotations.get(field_name, Any) for field_name in namedtuple_cls._fields}
1588 if typevars_map:
1589 annotations = {
1590 field_name: replace_types(annotation, typevars_map)
1591 for field_name, annotation in annotations.items()
1592 }
1594 schema = core_schema.named_tuple_schema(
1595 namedtuple_cls,
1596 [
1597 self._generate_namedtuple_field_schema(
1598 field_name,
1599 annotation,
1600 default=namedtuple_cls._field_defaults.get(field_name, Parameter.empty),
1601 )
1602 for field_name, annotation in annotations.items()
1603 ],
1604 ref=namedtuple_ref,
1605 )
1606 return self.defs.create_definition_reference_schema(schema)
1608 def _generate_namedtuple_field_schema(
1609 self,
1610 name: str,
1611 annotation: Any,
1612 default: Any = Parameter.empty,
1613 ) -> core_schema.NamedTupleField:
1614 """Generate the definition of a field in a named tuple."""
1615 FieldInfo = import_cached_field_info()
1617 if default is Parameter.empty:
1618 field = FieldInfo.from_annotation(annotation, _source=AnnotationSource.NAMED_TUPLE)
1619 else:
1620 field = FieldInfo.from_annotated_attribute(annotation, default, _source=AnnotationSource.NAMED_TUPLE)
1622 assert field.annotation is not None, 'field.annotation should not be None when generating a schema'
1623 update_field_from_config(self._config_wrapper, name, field)
1625 with self.field_name_stack.push(name):
1626 schema = self._apply_annotations(
1627 field.annotation,
1628 [field],
1629 # Because we pass `field` as metadata above (required for attributes relevant for
1630 # JSON Scheme generation), we need to ignore the potential warnings about `FieldInfo`
1631 # attributes that will not be used:
1632 check_unsupported_field_info_attributes=False,
1633 )
1635 if not field.is_required():
1636 schema = wrap_default(field, schema)
1638 return core_schema.named_tuple_field(
1639 name,
1640 schema,
1641 validation_alias=_convert_to_aliases(field.validation_alias),
1642 )
1644 def _generate_parameter_schema(
1645 self,
1646 name: str,
1647 annotation: type[Any],
1648 source: AnnotationSource,
1649 default: Any = Parameter.empty,
1650 mode: Literal['positional_only', 'positional_or_keyword', 'keyword_only'] | None = None,
1651 ) -> core_schema.ArgumentsParameter:
1652 """Generate the definition of a parameter in a function signature.
1654 This definition is meant to be used for the `'arguments'` core schema, which will be replaced
1655 in V3 by the `'arguments-v3`'.
1656 """
1657 FieldInfo = import_cached_field_info()
1659 if default is Parameter.empty:
1660 field = FieldInfo.from_annotation(annotation, _source=source)
1661 else:
1662 field = FieldInfo.from_annotated_attribute(annotation, default, _source=source)
1664 assert field.annotation is not None, 'field.annotation should not be None when generating a schema'
1665 update_field_from_config(self._config_wrapper, name, field)
1667 with self.field_name_stack.push(name):
1668 schema = self._apply_annotations(
1669 field.annotation,
1670 [field],
1671 # Because we pass `field` as metadata above (required for attributes relevant for
1672 # JSON Scheme generation), we need to ignore the potential warnings about `FieldInfo`
1673 # attributes that will not be used:
1674 check_unsupported_field_info_attributes=False,
1675 )
1677 if not field.is_required():
1678 schema = wrap_default(field, schema)
1680 parameter_schema = core_schema.arguments_parameter(
1681 name,
1682 schema,
1683 mode=mode,
1684 alias=_convert_to_aliases(field.validation_alias),
1685 )
1687 return parameter_schema
1689 def _generate_parameter_v3_schema(
1690 self,
1691 name: str,
1692 annotation: Any,
1693 source: AnnotationSource,
1694 mode: Literal[
1695 'positional_only',
1696 'positional_or_keyword',
1697 'keyword_only',
1698 'var_args',
1699 'var_kwargs_uniform',
1700 'var_kwargs_unpacked_typed_dict',
1701 ],
1702 default: Any = Parameter.empty,
1703 ) -> core_schema.ArgumentsV3Parameter:
1704 """Generate the definition of a parameter in a function signature.
1706 This definition is meant to be used for the `'arguments-v3'` core schema, which will replace
1707 the `'arguments`' schema in V3.
1708 """
1709 FieldInfo = import_cached_field_info()
1711 if default is Parameter.empty:
1712 field = FieldInfo.from_annotation(annotation, _source=source)
1713 else:
1714 field = FieldInfo.from_annotated_attribute(annotation, default, _source=source)
1715 update_field_from_config(self._config_wrapper, name, field)
1717 with self.field_name_stack.push(name):
1718 schema = self._apply_annotations(
1719 field.annotation,
1720 [field],
1721 # Because we pass `field` as metadata above (required for attributes relevant for
1722 # JSON Scheme generation), we need to ignore the potential warnings about `FieldInfo`
1723 # attributes that will not be used:
1724 check_unsupported_field_info_attributes=False,
1725 )
1727 if not field.is_required():
1728 schema = wrap_default(field, schema)
1730 parameter_schema = core_schema.arguments_v3_parameter(
1731 name=name,
1732 schema=schema,
1733 mode=mode,
1734 alias=_convert_to_aliases(field.validation_alias),
1735 )
1737 return parameter_schema
1739 def _tuple_schema(self, tuple_type: Any) -> core_schema.CoreSchema:
1740 """Generate schema for a Tuple, e.g. `tuple[int, str]` or `tuple[int, ...]`."""
1741 # TODO: do we really need to resolve type vars here?
1742 typevars_map = get_standard_typevars_map(tuple_type)
1743 params = self._get_args_resolving_forward_refs(tuple_type)
1745 if typevars_map and params:
1746 params = tuple(replace_types(param, typevars_map) for param in params)
1748 # NOTE: subtle difference: `tuple[()]` gives `params=()`, whereas `typing.Tuple[()]` gives `params=((),)`
1749 # This is only true for <3.11, on Python 3.11+ `typing.Tuple[()]` gives `params=()`
1750 if not params:
1751 if tuple_type in TUPLE_TYPES:
1752 return core_schema.tuple_schema([core_schema.any_schema()], variadic_item_index=0)
1753 else:
1754 # special case for `tuple[()]` which means `tuple[]` - an empty tuple
1755 return core_schema.tuple_schema([])
1756 elif params[-1] is Ellipsis:
1757 if len(params) == 2:
1758 return core_schema.tuple_schema([self.generate_schema(params[0])], variadic_item_index=0)
1759 else:
1760 # TODO: something like https://github.com/pydantic/pydantic/issues/5952
1761 raise ValueError('Variable tuples can only have one type')
1762 elif len(params) == 1 and params[0] == ():
1763 # special case for `tuple[()]` which means `tuple[]` - an empty tuple
1764 # NOTE: This conditional can be removed when we drop support for Python 3.10.
1765 return core_schema.tuple_schema([])
1766 else:
1767 return core_schema.tuple_schema([self.generate_schema(param) for param in params])
1769 def _type_schema(self) -> core_schema.CoreSchema:
1770 return core_schema.custom_error_schema(
1771 core_schema.is_instance_schema(type),
1772 custom_error_type='is_type',
1773 custom_error_message='Input should be a type',
1774 )
1776 def _zoneinfo_schema(self) -> core_schema.CoreSchema:
1777 """Generate schema for a zone_info.ZoneInfo object"""
1778 from ._validators import validate_str_is_valid_iana_tz
1780 metadata = {'pydantic_js_functions': [lambda _1, _2: {'type': 'string', 'format': 'zoneinfo'}]}
1781 return core_schema.no_info_plain_validator_function(
1782 validate_str_is_valid_iana_tz,
1783 serialization=core_schema.to_string_ser_schema(),
1784 metadata=metadata,
1785 )
1787 def _union_is_subclass_schema(self, union_type: Any) -> core_schema.CoreSchema:
1788 """Generate schema for `type[Union[X, ...]]`."""
1789 args = self._get_args_resolving_forward_refs(union_type, required=True)
1790 return core_schema.union_schema([self.generate_schema(type[args]) for args in args])
1792 def _subclass_schema(self, type_: Any) -> core_schema.CoreSchema:
1793 """Generate schema for a type, e.g. `type[int]`."""
1794 type_param = self._get_first_arg_or_any(type_)
1796 # Assume `type[Annotated[<typ>, ...]]` is equivalent to `type[<typ>]`:
1797 type_param = _typing_extra.annotated_type(type_param) or type_param
1799 if typing_objects.is_any(type_param):
1800 return self._type_schema()
1801 elif typing_objects.is_typealiastype(type_param):
1802 return self.generate_schema(type[type_param.__value__])
1803 elif typing_objects.is_typevar(type_param):
1804 if type_param.__bound__:
1805 if is_union_origin(get_origin(type_param.__bound__)):
1806 return self._union_is_subclass_schema(type_param.__bound__)
1807 return core_schema.is_subclass_schema(type_param.__bound__)
1808 elif type_param.__constraints__:
1809 return core_schema.union_schema([self.generate_schema(type[c]) for c in type_param.__constraints__])
1810 else:
1811 return self._type_schema()
1812 elif is_union_origin(get_origin(type_param)):
1813 return self._union_is_subclass_schema(type_param)
1814 else:
1815 if typing_objects.is_self(type_param):
1816 type_param = self._resolve_self_type(type_param)
1817 if _typing_extra.is_generic_alias(type_param):
1818 raise PydanticUserError(
1819 'Subscripting `type[]` with an already parametrized type is not supported. '
1820 f'Instead of using type[{type_param!r}], use type[{_repr.display_as_type(get_origin(type_param))}].',
1821 code=None,
1822 )
1823 if not inspect.isclass(type_param):
1824 # when using type[None], this doesn't type convert to type[NoneType], and None isn't a class
1825 # so we handle it manually here
1826 if type_param is None:
1827 return core_schema.is_subclass_schema(NoneType)
1828 raise TypeError(f'Expected a class, got {type_param!r}')
1829 return core_schema.is_subclass_schema(type_param)
1831 def _sequence_schema(self, items_type: Any) -> core_schema.CoreSchema:
1832 """Generate schema for a Sequence, e.g. `Sequence[int]`."""
1833 from ._serializers import serialize_sequence_via_list
1835 item_type_schema = self.generate_schema(items_type)
1836 list_schema = core_schema.list_schema(item_type_schema)
1838 json_schema = smart_deepcopy(list_schema)
1839 python_schema = core_schema.is_instance_schema(typing.Sequence, cls_repr='Sequence')
1840 if not typing_objects.is_any(items_type):
1841 from ._validators import sequence_validator
1843 python_schema = core_schema.chain_schema(
1844 [python_schema, core_schema.no_info_wrap_validator_function(sequence_validator, list_schema)],
1845 )
1847 serialization = core_schema.wrap_serializer_function_ser_schema(
1848 serialize_sequence_via_list, schema=item_type_schema, info_arg=True
1849 )
1850 return core_schema.json_or_python_schema(
1851 json_schema=json_schema, python_schema=python_schema, serialization=serialization
1852 )
1854 def _iterable_schema(self, type_: Any) -> core_schema.GeneratorSchema:
1855 """Generate a schema for an `Iterable`."""
1856 item_type = self._get_first_arg_or_any(type_)
1858 return core_schema.generator_schema(self.generate_schema(item_type))
1860 def _pattern_schema(self, pattern_type: Any) -> core_schema.CoreSchema:
1861 from . import _validators
1863 metadata = {'pydantic_js_functions': [lambda _1, _2: {'type': 'string', 'format': 'regex'}]}
1864 ser = core_schema.plain_serializer_function_ser_schema(
1865 attrgetter('pattern'), when_used='json', return_schema=core_schema.str_schema()
1866 )
1867 if pattern_type is typing.Pattern or pattern_type is re.Pattern:
1868 # bare type
1869 return core_schema.no_info_plain_validator_function(
1870 _validators.pattern_either_validator, serialization=ser, metadata=metadata
1871 )
1873 param = self._get_args_resolving_forward_refs(
1874 pattern_type,
1875 required=True,
1876 )[0]
1877 if param is str:
1878 return core_schema.no_info_plain_validator_function(
1879 _validators.pattern_str_validator, serialization=ser, metadata=metadata
1880 )
1881 elif param is bytes:
1882 return core_schema.no_info_plain_validator_function(
1883 _validators.pattern_bytes_validator, serialization=ser, metadata=metadata
1884 )
1885 else:
1886 raise PydanticSchemaGenerationError(f'Unable to generate pydantic-core schema for {pattern_type!r}.')
1888 def _hashable_schema(self) -> core_schema.CoreSchema:
1889 return core_schema.custom_error_schema(
1890 schema=core_schema.json_or_python_schema(
1891 json_schema=core_schema.chain_schema(
1892 [core_schema.any_schema(), core_schema.is_instance_schema(collections.abc.Hashable)]
1893 ),
1894 python_schema=core_schema.is_instance_schema(collections.abc.Hashable),
1895 ),
1896 custom_error_type='is_hashable',
1897 custom_error_message='Input should be hashable',
1898 )
1900 def _dataclass_schema(
1901 self, dataclass: type[StandardDataclass], origin: type[StandardDataclass] | None
1902 ) -> core_schema.CoreSchema:
1903 """Generate schema for a dataclass."""
1904 dataclass_ref = _type_refs.class_type_ref(dataclass, origin)
1905 with (
1906 self.model_type_stack.push(dataclass),
1907 self.defs.get_definition_ref_schema(ref=dataclass_ref) as maybe_schema,
1908 ):
1909 if maybe_schema is not None:
1910 return maybe_schema
1912 schema = dataclass.__dict__.get('__pydantic_core_schema__')
1913 if schema is not None and not isinstance(schema, MockCoreSchema):
1914 if schema['type'] == 'definitions':
1915 schema = self.defs.unpack_definitions(schema)
1916 ref = get_ref(schema)
1917 if ref:
1918 return self.defs.create_definition_reference_schema(schema)
1919 else:
1920 return schema
1922 typevars_map = get_standard_typevars_map(dataclass)
1923 if origin is not None:
1924 dataclass = origin
1926 # if (plain) dataclass doesn't have config, we use the parent's config, hence a default of `None`
1927 # (Pydantic dataclasses have an empty dict config by default).
1928 # see https://github.com/pydantic/pydantic/issues/10917
1929 config = getattr(dataclass, '__pydantic_config__', None)
1931 from ..dataclasses import is_pydantic_dataclass
1933 with self._ns_resolver.push(dataclass), self._config_wrapper_stack.push(config):
1934 if is_pydantic_dataclass(dataclass):
1935 if dataclass.__pydantic_fields_complete__():
1936 # Copy the field info instances to avoid mutating the `FieldInfo` instances
1937 # of the generic dataclass generic origin (e.g. `apply_typevars_map` below).
1938 # Note that we don't apply `deepcopy` on `__pydantic_fields__` because we
1939 # don't want to copy the `FieldInfo` attributes:
1940 fields = {
1941 f_name: copy(field_info) for f_name, field_info in dataclass.__pydantic_fields__.items()
1942 }
1943 if typevars_map:
1944 for field in fields.values():
1945 field.apply_typevars_map(typevars_map, *self._types_namespace)
1946 else:
1947 try:
1948 fields = rebuild_dataclass_fields(
1949 dataclass,
1950 config_wrapper=self._config_wrapper,
1951 ns_resolver=self._ns_resolver,
1952 typevars_map=typevars_map or {},
1953 )
1954 except NameError as e:
1955 raise PydanticUndefinedAnnotation.from_name_error(e) from e
1956 else:
1957 fields = collect_dataclass_fields(
1958 dataclass,
1959 typevars_map=typevars_map,
1960 config_wrapper=self._config_wrapper,
1961 )
1963 if self._config_wrapper.extra == 'allow':
1964 # disallow combination of init=False on a dataclass field and extra='allow' on a dataclass
1965 for field_name, field in fields.items():
1966 if field.init is False:
1967 raise PydanticUserError(
1968 f'Field {field_name} has `init=False` and dataclass has config setting `extra="allow"`. '
1969 f'This combination is not allowed.',
1970 code='dataclass-init-false-extra-allow',
1971 )
1973 decorators = dataclass.__dict__.get('__pydantic_decorators__')
1974 if decorators is None:
1975 decorators = DecoratorInfos.build(dataclass, replace_wrapped_methods=False)
1976 decorators.update_from_config(self._config_wrapper)
1977 # Move kw_only=False args to the start of the list, as this is how vanilla dataclasses work.
1978 # Note that when kw_only is missing or None, it is treated as equivalent to kw_only=True
1979 args = sorted(
1980 (self._generate_dc_field_schema(k, v, decorators) for k, v in fields.items()),
1981 key=lambda a: a.get('kw_only') is not False,
1982 )
1983 has_post_init = hasattr(dataclass, '__post_init__')
1984 has_slots = hasattr(dataclass, '__slots__')
1986 args_schema = core_schema.dataclass_args_schema(
1987 dataclass.__name__,
1988 args,
1989 computed_fields=[
1990 self._computed_field_schema(d, decorators.field_serializers)
1991 for d in decorators.computed_fields.values()
1992 ],
1993 collect_init_only=has_post_init,
1994 )
1996 inner_schema = apply_validators(args_schema, decorators.root_validators.values())
1998 model_validators = decorators.model_validators.values()
1999 inner_schema = apply_model_validators(inner_schema, model_validators, 'inner')
2001 core_config = self._config_wrapper.core_config(title=dataclass.__name__)
2003 dc_schema = core_schema.dataclass_schema(
2004 dataclass,
2005 inner_schema,
2006 generic_origin=origin,
2007 post_init=has_post_init,
2008 ref=dataclass_ref,
2009 fields=[field.name for field in dataclasses.fields(dataclass)],
2010 slots=has_slots,
2011 config=core_config,
2012 # we don't use a custom __setattr__ for dataclasses, so we must
2013 # pass along the frozen config setting to the pydantic-core schema
2014 frozen=self._config_wrapper_stack.tail.frozen,
2015 )
2016 schema = self._apply_model_serializers(dc_schema, decorators.model_serializers.values())
2017 schema = apply_model_validators(schema, model_validators, 'outer')
2018 return self.defs.create_definition_reference_schema(schema)
2020 def _call_schema(self, function: ValidateCallSupportedTypes) -> core_schema.CallSchema:
2021 """Generate schema for a Callable.
2023 TODO support functional validators once we support them in Config
2024 """
2025 arguments_schema = self._arguments_schema(function)
2027 return_schema: core_schema.CoreSchema | None = None
2028 config_wrapper = self._config_wrapper
2029 if config_wrapper.validate_return:
2030 sig = _typing_extra.signature_no_eval(function)
2031 return_hint = sig.return_annotation
2032 if return_hint is not sig.empty:
2033 globalns, localns = self._types_namespace
2034 type_hints = _typing_extra.get_function_type_hints(
2035 function, globalns=globalns, localns=localns, include_keys={'return'}
2036 )
2037 return_schema = self.generate_schema(type_hints['return'])
2039 return core_schema.call_schema(
2040 arguments_schema,
2041 function,
2042 return_schema=return_schema,
2043 )
2045 def _arguments_schema(
2046 self, function: ValidateCallSupportedTypes, parameters_callback: ParametersCallback | None = None
2047 ) -> core_schema.ArgumentsSchema:
2048 """Generate schema for a Signature."""
2049 mode_lookup: dict[_ParameterKind, Literal['positional_only', 'positional_or_keyword', 'keyword_only']] = {
2050 Parameter.POSITIONAL_ONLY: 'positional_only',
2051 Parameter.POSITIONAL_OR_KEYWORD: 'positional_or_keyword',
2052 Parameter.KEYWORD_ONLY: 'keyword_only',
2053 }
2055 sig = _typing_extra.signature_no_eval(function)
2056 globalns, localns = self._types_namespace
2057 type_hints = _typing_extra.get_function_type_hints(function, globalns=globalns, localns=localns)
2059 arguments_list: list[core_schema.ArgumentsParameter] = []
2060 var_args_schema: core_schema.CoreSchema | None = None
2061 var_kwargs_schema: core_schema.CoreSchema | None = None
2062 var_kwargs_mode: core_schema.VarKwargsMode | None = None
2064 core_config = self._config_wrapper.core_config(title=None)
2066 for i, (name, p) in enumerate(sig.parameters.items()):
2067 if p.annotation is sig.empty:
2068 annotation = typing.cast(Any, Any)
2069 else:
2070 annotation = type_hints[name]
2072 if parameters_callback is not None:
2073 result = parameters_callback(i, name, annotation)
2074 if result == 'skip':
2075 continue
2077 parameter_mode = mode_lookup.get(p.kind)
2078 if parameter_mode is not None:
2079 arg_schema = self._generate_parameter_schema(
2080 name, annotation, AnnotationSource.FUNCTION, p.default, parameter_mode
2081 )
2082 arguments_list.append(arg_schema)
2083 elif p.kind == Parameter.VAR_POSITIONAL:
2084 var_args_schema = self.generate_schema(annotation)
2085 else:
2086 assert p.kind == Parameter.VAR_KEYWORD, p.kind
2088 unpack_type = _typing_extra.unpack_type(annotation)
2089 if unpack_type is not None:
2090 origin = get_origin(unpack_type) or unpack_type
2091 if not is_typeddict(origin):
2092 raise PydanticUserError(
2093 f'Expected a `TypedDict` class inside `Unpack[...]`, got {unpack_type!r}',
2094 code='unpack-typed-dict',
2095 )
2096 non_pos_only_param_names = {
2097 name for name, p in sig.parameters.items() if p.kind != Parameter.POSITIONAL_ONLY
2098 }
2099 overlapping_params = non_pos_only_param_names.intersection(origin.__annotations__)
2100 if overlapping_params:
2101 raise PydanticUserError(
2102 f'Typed dictionary {origin.__name__!r} overlaps with parameter'
2103 f'{"s" if len(overlapping_params) >= 2 else ""} '
2104 f'{", ".join(repr(p) for p in sorted(overlapping_params))}',
2105 code='overlapping-unpack-typed-dict',
2106 )
2108 var_kwargs_mode = 'unpacked-typed-dict'
2109 var_kwargs_schema = self._typed_dict_schema(unpack_type, get_origin(unpack_type))
2110 else:
2111 var_kwargs_mode = 'uniform'
2112 var_kwargs_schema = self.generate_schema(annotation)
2114 return core_schema.arguments_schema(
2115 arguments_list,
2116 var_args_schema=var_args_schema,
2117 var_kwargs_mode=var_kwargs_mode,
2118 var_kwargs_schema=var_kwargs_schema,
2119 validate_by_name=core_config.get('validate_by_name'),
2120 )
2122 def _arguments_v3_schema(
2123 self, function: ValidateCallSupportedTypes, parameters_callback: ParametersCallback | None = None
2124 ) -> core_schema.ArgumentsV3Schema:
2125 mode_lookup: dict[
2126 _ParameterKind, Literal['positional_only', 'positional_or_keyword', 'var_args', 'keyword_only']
2127 ] = {
2128 Parameter.POSITIONAL_ONLY: 'positional_only',
2129 Parameter.POSITIONAL_OR_KEYWORD: 'positional_or_keyword',
2130 Parameter.VAR_POSITIONAL: 'var_args',
2131 Parameter.KEYWORD_ONLY: 'keyword_only',
2132 }
2134 sig = _typing_extra.signature_no_eval(function)
2135 globalns, localns = self._types_namespace
2136 type_hints = _typing_extra.get_function_type_hints(function, globalns=globalns, localns=localns)
2138 parameters_list: list[core_schema.ArgumentsV3Parameter] = []
2139 core_config = self._config_wrapper.core_config(title=None)
2141 for i, (name, p) in enumerate(sig.parameters.items()):
2142 if parameters_callback is not None:
2143 result = parameters_callback(i, name, p.annotation)
2144 if result == 'skip':
2145 continue
2147 if p.annotation is Parameter.empty:
2148 annotation = typing.cast(Any, Any)
2149 else:
2150 annotation = type_hints[name]
2152 parameter_mode = mode_lookup.get(p.kind)
2153 if parameter_mode is None:
2154 assert p.kind == Parameter.VAR_KEYWORD, p.kind
2156 unpack_type = _typing_extra.unpack_type(annotation)
2157 if unpack_type is not None:
2158 origin = get_origin(unpack_type) or unpack_type
2159 if not is_typeddict(origin):
2160 raise PydanticUserError(
2161 f'Expected a `TypedDict` class inside `Unpack[...]`, got {unpack_type!r}',
2162 code='unpack-typed-dict',
2163 )
2164 non_pos_only_param_names = {
2165 name for name, p in sig.parameters.items() if p.kind != Parameter.POSITIONAL_ONLY
2166 }
2167 overlapping_params = non_pos_only_param_names.intersection(origin.__annotations__)
2168 if overlapping_params:
2169 raise PydanticUserError(
2170 f'Typed dictionary {origin.__name__!r} overlaps with parameter'
2171 f'{"s" if len(overlapping_params) >= 2 else ""} '
2172 f'{", ".join(repr(p) for p in sorted(overlapping_params))}',
2173 code='overlapping-unpack-typed-dict',
2174 )
2175 parameter_mode = 'var_kwargs_unpacked_typed_dict'
2176 annotation = unpack_type
2177 else:
2178 parameter_mode = 'var_kwargs_uniform'
2180 parameters_list.append(
2181 self._generate_parameter_v3_schema(
2182 name, annotation, AnnotationSource.FUNCTION, parameter_mode, default=p.default
2183 )
2184 )
2186 return core_schema.arguments_v3_schema(
2187 parameters_list,
2188 validate_by_name=core_config.get('validate_by_name'),
2189 )
2191 def _unsubstituted_typevar_schema(self, typevar: typing.TypeVar) -> core_schema.CoreSchema:
2192 try:
2193 has_default = typevar.has_default() # pyright: ignore[reportAttributeAccessIssue]
2194 except AttributeError:
2195 # Happens if using `typing.TypeVar` (and not `typing_extensions`) on Python < 3.13
2196 pass
2197 else:
2198 if has_default:
2199 return self.generate_schema(typevar.__default__) # pyright: ignore[reportAttributeAccessIssue]
2201 if constraints := typevar.__constraints__:
2202 return self._union_schema(typing.Union[constraints]) # noqa: UP007
2204 if bound := typevar.__bound__:
2205 schema = self.generate_schema(bound)
2206 schema['serialization'] = core_schema.simple_ser_schema('any')
2207 return schema
2209 return core_schema.any_schema()
2211 def _computed_field_schema(
2212 self,
2213 d: Decorator[ComputedFieldInfo],
2214 field_serializers: dict[str, Decorator[FieldSerializerDecoratorInfo]],
2215 ) -> core_schema.ComputedField:
2216 if d.info.return_type is not PydanticUndefined:
2217 return_type = d.info.return_type
2218 else:
2219 try:
2220 # Do not pass in globals as the function could be defined in a different module.
2221 # Instead, let `get_callable_return_type` infer the globals to use, but still pass
2222 # in locals that may contain a parent/rebuild namespace:
2223 return_type = _decorators.get_callable_return_type(d.func, localns=self._types_namespace.locals)
2224 except NameError as e:
2225 raise PydanticUndefinedAnnotation.from_name_error(e) from e
2226 if return_type is PydanticUndefined:
2227 raise PydanticUserError(
2228 'Computed field is missing return type annotation or specifying `return_type`'
2229 ' to the `@computed_field` decorator (e.g. `@computed_field(return_type=int | str)`)',
2230 code='model-field-missing-annotation',
2231 )
2233 return_type = replace_types(return_type, self._typevars_map)
2234 # Create a new ComputedFieldInfo so that different type parametrizations of the same
2235 # generic model's computed field can have different return types.
2236 d.info = dataclasses.replace(d.info, return_type=return_type)
2237 return_type_schema = self.generate_schema(return_type)
2238 # Apply serializers to computed field if there exist
2239 return_type_schema = self._apply_field_serializers(
2240 return_type_schema,
2241 filter_field_decorator_info_by_field(field_serializers.values(), d.cls_var_name),
2242 )
2244 pydantic_js_updates, pydantic_js_extra = _extract_json_schema_info_from_field_info(d.info)
2245 core_metadata: dict[str, Any] = {}
2246 update_core_metadata(
2247 core_metadata,
2248 pydantic_js_updates={'readOnly': True, **(pydantic_js_updates if pydantic_js_updates else {})},
2249 pydantic_js_extra=pydantic_js_extra,
2250 )
2251 exclude_if = d.info.exclude_if
2252 # TODO: Should we support exclude_if from annotations?
2253 return core_schema.computed_field(
2254 d.cls_var_name,
2255 return_schema=return_type_schema,
2256 alias=d.info.alias,
2257 serialization_exclude_if=exclude_if,
2258 metadata=core_metadata,
2259 )
2261 def _annotated_schema(self, annotated_type: Any) -> core_schema.CoreSchema:
2262 """Generate schema for an Annotated type, e.g. `Annotated[int, Field(...)]` or `Annotated[int, Gt(0)]`."""
2263 FieldInfo = import_cached_field_info()
2264 source_type, *annotations = self._get_args_resolving_forward_refs(
2265 annotated_type,
2266 required=True,
2267 )
2268 schema = self._apply_annotations(source_type, annotations)
2269 # put the default validator last so that TypeAdapter.get_default_value() works
2270 # even if there are function validators involved
2271 for annotation in annotations:
2272 if isinstance(annotation, FieldInfo):
2273 schema = wrap_default(annotation, schema)
2274 return schema
2276 def _apply_annotations(
2277 self,
2278 source_type: Any,
2279 annotations: list[Any],
2280 transform_inner_schema: Callable[[CoreSchema], CoreSchema] = lambda x: x,
2281 check_unsupported_field_info_attributes: bool = True,
2282 ) -> CoreSchema:
2283 """Apply arguments from `Annotated` or from `FieldInfo` to a schema.
2285 This gets called by `GenerateSchema._annotated_schema` but differs from it in that it does
2286 not expect `source_type` to be an `Annotated` object, it expects it to be the first argument of that
2287 (in other words, `GenerateSchema._annotated_schema` just unpacks `Annotated`, this process it).
2288 """
2289 annotations = list(_known_annotated_metadata.expand_grouped_metadata(annotations))
2291 pydantic_js_annotation_functions: list[GetJsonSchemaFunction] = []
2293 def inner_handler(obj: Any) -> CoreSchema:
2294 origin = get_origin(obj)
2295 schema: CoreSchema | None = None
2296 if origin is None:
2297 # If the type has an origin, it is guaranteed to *not* be an ordinary class that could have
2298 # a `__get_pydantic_core_schema__()` method:
2299 schema = self._generate_schema_from_get_schema_method(obj, source_type)
2301 if schema is None:
2302 schema = self._generate_schema_inner(obj, origin)
2304 # `__get_pydantic_json_schema__()` is checked on the origin if it exists:
2305 metadata_js_function = _extract_get_pydantic_json_schema(origin if origin is not None else obj)
2306 if metadata_js_function is not None:
2307 metadata_schema = resolve_original_schema(schema, self.defs)
2308 if metadata_schema is not None:
2309 self._add_js_function(metadata_schema, metadata_js_function)
2310 return transform_inner_schema(schema)
2312 get_inner_schema = CallbackGetCoreSchemaHandler(inner_handler, self)
2314 for annotation in annotations:
2315 if annotation is None:
2316 continue
2317 get_inner_schema = self._get_wrapped_inner_schema(
2318 get_inner_schema,
2319 annotation,
2320 pydantic_js_annotation_functions,
2321 check_unsupported_field_info_attributes=check_unsupported_field_info_attributes,
2322 )
2324 schema = get_inner_schema(source_type)
2325 if pydantic_js_annotation_functions:
2326 core_metadata = schema.setdefault('metadata', {})
2327 update_core_metadata(core_metadata, pydantic_js_annotation_functions=pydantic_js_annotation_functions)
2328 return _add_custom_serialization_from_json_encoders(self._config_wrapper.json_encoders, source_type, schema)
2330 def _apply_single_annotation(
2331 self,
2332 schema: core_schema.CoreSchema,
2333 metadata: Any,
2334 check_unsupported_field_info_attributes: bool = True,
2335 ) -> core_schema.CoreSchema:
2336 FieldInfo = import_cached_field_info()
2338 if isinstance(metadata, FieldInfo):
2339 if (
2340 check_unsupported_field_info_attributes
2341 # HACK: we don't want to emit the warning for `FieldInfo` subclasses, because FastAPI does weird manipulations
2342 # with its subclasses and their annotations:
2343 and type(metadata) is FieldInfo
2344 ):
2345 for attr, value in (unsupported_attributes := self._get_unsupported_field_info_attributes(metadata)):
2346 warnings.warn(
2347 f'The {attr!r} attribute with value {value!r} was provided to the `Field()` function, '
2348 f'which has no effect in the context it was used. {attr!r} is field-specific metadata, '
2349 'and can only be attached to a model field using `Annotated` metadata or by assignment. '
2350 'This may have happened because an `Annotated` type alias using the `type` statement was '
2351 'used, or if the `Field()` function was attached to a single member of a union type.',
2352 category=UnsupportedFieldAttributeWarning,
2353 )
2355 if (
2356 metadata.default_factory_takes_validated_data
2357 and self.model_type_stack.get() is None
2358 and 'defaut_factory' not in unsupported_attributes
2359 ):
2360 warnings.warn(
2361 "A 'default_factory' taking validated data as an argument was provided to the `Field()` function, "
2362 'but no validated data is available in the context it was used.',
2363 category=UnsupportedFieldAttributeWarning,
2364 )
2366 for field_metadata in metadata.metadata:
2367 schema = self._apply_single_annotation(schema, field_metadata)
2369 if metadata.discriminator is not None:
2370 schema = self._apply_discriminator_to_union(schema, metadata.discriminator)
2371 return schema
2373 if schema['type'] == 'nullable':
2374 # for nullable schemas, metadata is automatically applied to the inner schema
2375 inner = schema.get('schema', core_schema.any_schema())
2376 inner = self._apply_single_annotation(inner, metadata)
2377 if inner:
2378 schema['schema'] = inner
2379 return schema
2381 if schema['type'] == 'missing-sentinel' and (inner := schema.get('schema')) is not None:
2382 # Same behavior as for nullable schemas: metadata is automatically applied to the inner schema
2383 inner = self._apply_single_annotation(inner, metadata)
2384 if inner:
2385 schema['schema'] = inner
2386 return schema
2388 original_schema = schema
2389 ref = schema.get('ref')
2390 if ref is not None:
2391 schema = schema.copy()
2392 new_ref = ref + f'_{repr(metadata)}'
2393 if (existing := self.defs.get_schema_from_ref(new_ref)) is not None:
2394 return existing
2395 schema['ref'] = new_ref # pyright: ignore[reportGeneralTypeIssues]
2396 elif schema['type'] == 'definition-ref':
2397 ref = schema['schema_ref']
2398 if (referenced_schema := self.defs.get_schema_from_ref(ref)) is not None:
2399 schema = referenced_schema.copy()
2400 new_ref = ref + f'_{repr(metadata)}'
2401 if (existing := self.defs.get_schema_from_ref(new_ref)) is not None:
2402 return existing
2403 schema['ref'] = new_ref # pyright: ignore[reportGeneralTypeIssues]
2405 maybe_updated_schema = _known_annotated_metadata.apply_known_metadata(metadata, schema)
2407 if maybe_updated_schema is not None:
2408 return maybe_updated_schema
2409 return original_schema
2411 def _apply_single_annotation_json_schema(
2412 self, schema: core_schema.CoreSchema, metadata: Any
2413 ) -> core_schema.CoreSchema:
2414 FieldInfo = import_cached_field_info()
2416 if isinstance(metadata, FieldInfo):
2417 for field_metadata in metadata.metadata:
2418 schema = self._apply_single_annotation_json_schema(schema, field_metadata)
2420 pydantic_js_updates, pydantic_js_extra = _extract_json_schema_info_from_field_info(metadata)
2421 core_metadata = schema.setdefault('metadata', {})
2422 update_core_metadata(
2423 core_metadata, pydantic_js_updates=pydantic_js_updates, pydantic_js_extra=pydantic_js_extra
2424 )
2425 return schema
2427 def _get_unsupported_field_info_attributes(self, field_info: FieldInfo) -> list[tuple[str, Any]]:
2428 """Get the list of unsupported `FieldInfo` attributes when not directly used in `Annotated` for field annotations."""
2429 unused_metadata: list[tuple[str, Any]] = []
2430 for unused_metadata_name, unset_value in UNSUPPORTED_STANDALONE_FIELDINFO_ATTRIBUTES:
2431 if (
2432 (unused_metadata_value := getattr(field_info, unused_metadata_name)) is not unset_value
2433 # `default` and `default_factory` can still be used with a type adapter, so only include them
2434 # if used with a model-like class:
2435 and (
2436 unused_metadata_name not in ('default', 'default_factory')
2437 or self.model_type_stack.get() is not None
2438 )
2439 # Setting `alias` will set `validation/serialization_alias` as well, so we want to avoid duplicate warnings:
2440 and (
2441 unused_metadata_name not in ('validation_alias', 'serialization_alias')
2442 or 'alias' not in field_info._attributes_set
2443 )
2444 ):
2445 unused_metadata.append((unused_metadata_name, unused_metadata_value))
2447 return unused_metadata
2449 def _get_wrapped_inner_schema(
2450 self,
2451 get_inner_schema: GetCoreSchemaHandler,
2452 annotation: Any,
2453 pydantic_js_annotation_functions: list[GetJsonSchemaFunction],
2454 check_unsupported_field_info_attributes: bool = False,
2455 ) -> CallbackGetCoreSchemaHandler:
2456 annotation_get_schema: GetCoreSchemaFunction | None = getattr(annotation, '__get_pydantic_core_schema__', None)
2458 def new_handler(source: Any) -> core_schema.CoreSchema:
2459 if annotation_get_schema is not None:
2460 schema = annotation_get_schema(source, get_inner_schema)
2461 else:
2462 schema = get_inner_schema(source)
2463 schema = self._apply_single_annotation(
2464 schema,
2465 annotation,
2466 check_unsupported_field_info_attributes=check_unsupported_field_info_attributes,
2467 )
2468 schema = self._apply_single_annotation_json_schema(schema, annotation)
2470 metadata_js_function = _extract_get_pydantic_json_schema(annotation)
2471 if metadata_js_function is not None:
2472 pydantic_js_annotation_functions.append(metadata_js_function)
2473 return schema
2475 return CallbackGetCoreSchemaHandler(new_handler, self)
2477 def _apply_field_serializers(
2478 self,
2479 schema: core_schema.CoreSchema,
2480 serializers: list[Decorator[FieldSerializerDecoratorInfo]],
2481 ) -> core_schema.CoreSchema:
2482 """Apply field serializers to a schema."""
2483 if serializers:
2484 schema = copy(schema)
2485 if schema['type'] == 'definitions':
2486 inner_schema = schema['schema']
2487 schema['schema'] = self._apply_field_serializers(inner_schema, serializers)
2488 return schema
2489 elif 'ref' in schema:
2490 schema = self.defs.create_definition_reference_schema(schema)
2492 # use the last serializer to make it easy to override a serializer set on a parent model
2493 serializer = serializers[-1]
2494 is_field_serializer, info_arg = inspect_field_serializer(serializer.func, serializer.info.mode)
2496 if serializer.info.return_type is not PydanticUndefined:
2497 return_type = serializer.info.return_type
2498 else:
2499 try:
2500 # Do not pass in globals as the function could be defined in a different module.
2501 # Instead, let `get_callable_return_type` infer the globals to use, but still pass
2502 # in locals that may contain a parent/rebuild namespace:
2503 return_type = _decorators.get_callable_return_type(
2504 serializer.func, localns=self._types_namespace.locals
2505 )
2506 except NameError as e:
2507 raise PydanticUndefinedAnnotation.from_name_error(e) from e
2509 if return_type is PydanticUndefined:
2510 return_schema = None
2511 else:
2512 return_schema = self.generate_schema(return_type)
2514 if serializer.info.mode == 'wrap':
2515 schema['serialization'] = core_schema.wrap_serializer_function_ser_schema(
2516 serializer.func,
2517 is_field_serializer=is_field_serializer,
2518 info_arg=info_arg,
2519 return_schema=return_schema,
2520 when_used=serializer.info.when_used,
2521 )
2522 else:
2523 assert serializer.info.mode == 'plain'
2524 schema['serialization'] = core_schema.plain_serializer_function_ser_schema(
2525 serializer.func,
2526 is_field_serializer=is_field_serializer,
2527 info_arg=info_arg,
2528 return_schema=return_schema,
2529 when_used=serializer.info.when_used,
2530 )
2531 return schema
2533 def _apply_model_serializers(
2534 self, schema: core_schema.CoreSchema, serializers: Iterable[Decorator[ModelSerializerDecoratorInfo]]
2535 ) -> core_schema.CoreSchema:
2536 """Apply model serializers to a schema."""
2537 ref: str | None = schema.pop('ref', None) # type: ignore
2538 if serializers:
2539 serializer = list(serializers)[-1]
2540 info_arg = inspect_model_serializer(serializer.func, serializer.info.mode)
2542 if serializer.info.return_type is not PydanticUndefined:
2543 return_type = serializer.info.return_type
2544 else:
2545 try:
2546 # Do not pass in globals as the function could be defined in a different module.
2547 # Instead, let `get_callable_return_type` infer the globals to use, but still pass
2548 # in locals that may contain a parent/rebuild namespace:
2549 return_type = _decorators.get_callable_return_type(
2550 serializer.func, localns=self._types_namespace.locals
2551 )
2552 except NameError as e:
2553 raise PydanticUndefinedAnnotation.from_name_error(e) from e
2555 if return_type is PydanticUndefined:
2556 return_schema = None
2557 else:
2558 return_schema = self.generate_schema(return_type)
2560 if serializer.info.mode == 'wrap':
2561 ser_schema: core_schema.SerSchema = core_schema.wrap_serializer_function_ser_schema(
2562 serializer.func,
2563 info_arg=info_arg,
2564 return_schema=return_schema,
2565 when_used=serializer.info.when_used,
2566 )
2567 else:
2568 # plain
2569 ser_schema = core_schema.plain_serializer_function_ser_schema(
2570 serializer.func,
2571 info_arg=info_arg,
2572 return_schema=return_schema,
2573 when_used=serializer.info.when_used,
2574 )
2575 schema['serialization'] = ser_schema
2576 if ref:
2577 schema['ref'] = ref # type: ignore
2578 return schema
2581_VALIDATOR_F_MATCH: Mapping[
2582 tuple[FieldValidatorModes, Literal['no-info', 'with-info']],
2583 Callable[[Callable[..., Any], core_schema.CoreSchema], core_schema.CoreSchema],
2584] = {
2585 ('before', 'no-info'): lambda f, schema: core_schema.no_info_before_validator_function(f, schema),
2586 ('after', 'no-info'): lambda f, schema: core_schema.no_info_after_validator_function(f, schema),
2587 ('plain', 'no-info'): lambda f, _: core_schema.no_info_plain_validator_function(f),
2588 ('wrap', 'no-info'): lambda f, schema: core_schema.no_info_wrap_validator_function(f, schema),
2589 ('before', 'with-info'): lambda f, schema: core_schema.with_info_before_validator_function(f, schema),
2590 ('after', 'with-info'): lambda f, schema: core_schema.with_info_after_validator_function(f, schema),
2591 ('plain', 'with-info'): lambda f, _: core_schema.with_info_plain_validator_function(f),
2592 ('wrap', 'with-info'): lambda f, schema: core_schema.with_info_wrap_validator_function(f, schema),
2593}
2596# TODO V3: this function is only used for deprecated decorators. It should
2597# be removed once we drop support for those.
2598def apply_validators(
2599 schema: core_schema.CoreSchema,
2600 validators: Iterable[Decorator[RootValidatorDecoratorInfo]]
2601 | Iterable[Decorator[ValidatorDecoratorInfo]]
2602 | Iterable[Decorator[FieldValidatorDecoratorInfo]],
2603) -> core_schema.CoreSchema:
2604 """Apply validators to a schema.
2606 Args:
2607 schema: The schema to apply validators on.
2608 validators: An iterable of validators.
2610 Returns:
2611 The updated schema.
2612 """
2613 for validator in validators:
2614 # Actually, type could be 'field' or 'model', but this is only used for deprecated
2615 # decorators, so let's not worry about it.
2616 info_arg = inspect_validator(validator.func, mode=validator.info.mode, type='field')
2617 val_type = 'with-info' if info_arg else 'no-info'
2619 schema = _VALIDATOR_F_MATCH[(validator.info.mode, val_type)](validator.func, schema)
2620 return schema
2623def _validators_require_validate_default(validators: Iterable[Decorator[ValidatorDecoratorInfo]]) -> bool:
2624 """In v1, if any of the validators for a field had `always=True`, the default value would be validated.
2626 This serves as an auxiliary function for re-implementing that logic, by looping over a provided
2627 collection of (v1-style) ValidatorDecoratorInfo's and checking if any of them have `always=True`.
2629 We should be able to drop this function and the associated logic calling it once we drop support
2630 for v1-style validator decorators. (Or we can extend it and keep it if we add something equivalent
2631 to the v1-validator `always` kwarg to `field_validator`.)
2632 """
2633 for validator in validators:
2634 if validator.info.always:
2635 return True
2636 return False
2639def _convert_to_aliases(
2640 alias: str | AliasChoices | AliasPath | None,
2641) -> str | list[str | int] | list[list[str | int]] | None:
2642 if isinstance(alias, (AliasChoices, AliasPath)):
2643 return alias.convert_to_aliases()
2644 else:
2645 return alias
2648def apply_model_validators(
2649 schema: core_schema.CoreSchema,
2650 validators: Iterable[Decorator[ModelValidatorDecoratorInfo]],
2651 mode: Literal['inner', 'outer', 'all'],
2652) -> core_schema.CoreSchema:
2653 """Apply model validators to a schema.
2655 If mode == 'inner', only "before" validators are applied
2656 If mode == 'outer', validators other than "before" are applied
2657 If mode == 'all', all validators are applied
2659 Args:
2660 schema: The schema to apply validators on.
2661 validators: An iterable of validators.
2662 mode: The validator mode.
2664 Returns:
2665 The updated schema.
2666 """
2667 ref: str | None = schema.pop('ref', None) # type: ignore
2668 for validator in validators:
2669 if mode == 'inner' and validator.info.mode != 'before':
2670 continue
2671 if mode == 'outer' and validator.info.mode == 'before':
2672 continue
2673 info_arg = inspect_validator(validator.func, mode=validator.info.mode, type='model')
2674 if validator.info.mode == 'wrap':
2675 if info_arg:
2676 schema = core_schema.with_info_wrap_validator_function(function=validator.func, schema=schema)
2677 else:
2678 schema = core_schema.no_info_wrap_validator_function(function=validator.func, schema=schema)
2679 elif validator.info.mode == 'before':
2680 if info_arg:
2681 schema = core_schema.with_info_before_validator_function(function=validator.func, schema=schema)
2682 else:
2683 schema = core_schema.no_info_before_validator_function(function=validator.func, schema=schema)
2684 else:
2685 assert validator.info.mode == 'after'
2686 if info_arg:
2687 schema = core_schema.with_info_after_validator_function(function=validator.func, schema=schema)
2688 else:
2689 schema = core_schema.no_info_after_validator_function(function=validator.func, schema=schema)
2690 if ref:
2691 schema['ref'] = ref # type: ignore
2692 return schema
2695def wrap_default(field_info: FieldInfo, schema: core_schema.CoreSchema) -> core_schema.CoreSchema:
2696 """Wrap schema with default schema if default value or `default_factory` are available.
2698 Args:
2699 field_info: The field info object.
2700 schema: The schema to apply default on.
2702 Returns:
2703 Updated schema by default value or `default_factory`.
2704 """
2705 if field_info.default_factory:
2706 return core_schema.with_default_schema(
2707 schema,
2708 default_factory=field_info.default_factory,
2709 default_factory_takes_data=takes_validated_data_argument(field_info.default_factory),
2710 validate_default=field_info.validate_default,
2711 )
2712 elif field_info.default is not PydanticUndefined:
2713 return core_schema.with_default_schema(
2714 schema, default=field_info.default, validate_default=field_info.validate_default
2715 )
2716 else:
2717 return schema
2720def _extract_get_pydantic_json_schema(tp: Any) -> GetJsonSchemaFunction | None:
2721 """Extract `__get_pydantic_json_schema__` from a type, handling the deprecated `__modify_schema__`."""
2722 js_modify_function = getattr(tp, '__get_pydantic_json_schema__', None)
2724 if hasattr(tp, '__modify_schema__'):
2725 BaseModel = import_cached_base_model()
2727 has_custom_v2_modify_js_func = (
2728 js_modify_function is not None
2729 and BaseModel.__get_pydantic_json_schema__.__func__ # type: ignore
2730 not in (js_modify_function, getattr(js_modify_function, '__func__', None))
2731 )
2733 if not has_custom_v2_modify_js_func:
2734 cls_name = getattr(tp, '__name__', None)
2735 raise PydanticUserError(
2736 f'The `__modify_schema__` method is not supported in Pydantic v2. '
2737 f'Use `__get_pydantic_json_schema__` instead{f" in class `{cls_name}`" if cls_name else ""}.',
2738 code='custom-json-schema',
2739 )
2741 if js_modify_function is None:
2742 return None
2744 return js_modify_function
2747def resolve_original_schema(schema: CoreSchema, definitions: _Definitions) -> CoreSchema | None:
2748 if schema['type'] == 'definition-ref':
2749 return definitions.get_schema_from_ref(schema['schema_ref'])
2750 elif schema['type'] == 'definitions':
2751 return schema['schema']
2752 else:
2753 return schema
2756def _inlining_behavior(
2757 def_ref: core_schema.DefinitionReferenceSchema,
2758) -> Literal['inline', 'keep', 'preserve_metadata']:
2759 """Determine the inlining behavior of the `'definition-ref'` schema.
2761 - If no `'serialization'` schema and no metadata is attached, the schema can safely be inlined.
2762 - If it has metadata but only related to the deferred discriminator application, it can be inlined
2763 provided that such metadata is kept.
2764 - Otherwise, the schema should not be inlined. Doing so would remove the `'serialization'` schema or metadata.
2765 """
2766 if 'serialization' in def_ref:
2767 return 'keep'
2768 metadata = def_ref.get('metadata')
2769 if not metadata:
2770 return 'inline'
2771 if len(metadata) == 1 and 'pydantic_internal_union_discriminator' in metadata:
2772 return 'preserve_metadata'
2773 return 'keep'
2776class _Definitions:
2777 """Keeps track of references and definitions."""
2779 _recursively_seen: set[str]
2780 """A set of recursively seen references.
2782 When a referenceable type is encountered, the `get_schema_or_ref` context manager is
2783 entered to compute the reference. If the type references itself by some way (e.g. for
2784 a dataclass a Pydantic model, the class can be referenced as a field annotation),
2785 entering the context manager again will yield a `'definition-ref'` schema that should
2786 short-circuit the normal generation process, as the reference was already in this set.
2787 """
2789 _definitions: dict[str, core_schema.CoreSchema]
2790 """A mapping of references to their corresponding schema.
2792 When a schema for a referenceable type is generated, it is stored in this mapping. If the
2793 same type is encountered again, the reference is yielded by the `get_schema_or_ref` context
2794 manager.
2795 """
2797 def __init__(self) -> None:
2798 self._recursively_seen = set()
2799 self._definitions = {}
2801 @contextmanager
2802 def get_definition_ref_schema(self, *, ref: str) -> Generator[core_schema.DefinitionReferenceSchema | None]:
2803 """Get a `'definition-ref'` schema for `ref`, if the reference was already encountered (or if we are in a cycle).
2804 Get a definition for `tp` if one exists.
2806 This should be called for any type that can be identified by reference.
2807 This includes any recursive types.
2809 At present the following types can be named/recursive:
2811 - Pydantic model
2812 - Pydantic and stdlib dataclasses
2813 - Typed dictionaries
2814 - Named tuples
2815 - `TypeAliasType` instances
2816 - Enums
2817 """
2818 # return the reference if we're either (1) in a cycle or (2) it the reference was already encountered:
2819 if ref in self._recursively_seen or ref in self._definitions:
2820 yield core_schema.definition_reference_schema(ref)
2821 else:
2822 self._recursively_seen.add(ref)
2823 try:
2824 yield None
2825 finally:
2826 self._recursively_seen.discard(ref)
2828 def get_schema_from_ref(self, ref: str) -> CoreSchema | None:
2829 """Resolve the schema from the given reference."""
2830 return self._definitions.get(ref)
2832 def create_definition_reference_schema(self, schema: CoreSchema) -> core_schema.DefinitionReferenceSchema:
2833 """Store the schema as a definition and return a `'definition-reference'` schema pointing to it.
2835 The schema must have a reference attached to it.
2836 """
2837 ref = schema['ref'] # pyright: ignore
2838 self._definitions[ref] = schema
2839 return core_schema.definition_reference_schema(ref)
2841 def unpack_definitions(self, schema: core_schema.DefinitionsSchema) -> CoreSchema:
2842 """Store the definitions of the `'definitions'` core schema and return the inner core schema."""
2843 for def_schema in schema['definitions']:
2844 self._definitions[def_schema['ref']] = def_schema # pyright: ignore
2845 return schema['schema']
2847 def finalize_schema(self, schema: CoreSchema) -> CoreSchema:
2848 """Finalize the core schema.
2850 This traverses the core schema and referenced definitions, replaces `'definition-ref'` schemas
2851 by the referenced definition if possible, and applies deferred discriminators.
2852 """
2853 definitions = self._definitions
2854 try:
2855 gather_result = gather_schemas_for_cleaning(
2856 schema,
2857 definitions=definitions,
2858 )
2859 except MissingDefinitionError as e:
2860 raise InvalidSchemaError from e
2862 remaining_defs: dict[str, CoreSchema] = {}
2864 # References that will be kept as actual definitions (i.e. not inlined). When resolving an
2865 # "intermediate" `'definition-ref'` schema (see `_resolve_definition()`), we must not follow
2866 # the reference chain through one of these definitions:
2867 kept_refs = {
2868 ref
2869 for ref, inlinable_def_ref in gather_result['collected_references'].items()
2870 if inlinable_def_ref is None or _inlining_behavior(inlinable_def_ref) == 'keep'
2871 }
2873 # Note: this logic doesn't play well when core schemas with deferred discriminator metadata
2874 # and references are encountered. See the `test_deferred_discriminated_union_and_references()` test.
2875 for ref, inlinable_def_ref in gather_result['collected_references'].items():
2876 if inlinable_def_ref is not None and (inlining_behavior := _inlining_behavior(inlinable_def_ref)) != 'keep':
2877 if inlining_behavior == 'inline':
2878 # `ref` was encountered, and only once:
2879 # - `inlinable_def_ref` is a `'definition-ref'` schema and is guaranteed to be
2880 # the only one. Transform it into the definition it points to.
2881 # - Do not store the definition in the `remaining_defs`.
2882 inlinable_def_ref.clear() # pyright: ignore[reportAttributeAccessIssue]
2883 inlinable_def_ref.update(self._resolve_definition(ref, definitions, kept_refs)) # pyright: ignore
2884 elif inlining_behavior == 'preserve_metadata':
2885 # `ref` was encountered, and only once, but contains discriminator metadata.
2886 # We will do the same thing as if `inlining_behavior` was `'inline'`, but make
2887 # sure to keep the metadata for the deferred discriminator application logic below.
2888 meta = inlinable_def_ref.pop('metadata')
2889 inlinable_def_ref.clear() # pyright: ignore[reportAttributeAccessIssue]
2890 inlinable_def_ref.update(self._resolve_definition(ref, definitions, kept_refs)) # pyright: ignore
2891 inlinable_def_ref['metadata'] = meta
2892 else:
2893 # `ref` was encountered, at least two times (or only once, but with metadata or a serialization schema):
2894 # - Do not inline the `'definition-ref'` schemas (they are not provided in the gather result anyway).
2895 # - Store the definition in the `remaining_defs`
2896 remaining_defs[ref] = self._resolve_definition(ref, definitions, kept_refs)
2898 for cs in gather_result['deferred_discriminator_schemas']:
2899 discriminator: str | None = cs['metadata'].pop('pydantic_internal_union_discriminator', None) # pyright: ignore[reportTypedDictNotRequiredAccess]
2900 if discriminator is None:
2901 # This can happen in rare scenarios, when a deferred schema is present multiple times in the
2902 # gather result (e.g. when using the `Sequence` type -- see `test_sequence_discriminated_union()`).
2903 # In this case, a previous loop iteration applied the discriminator and so we can just skip it here.
2904 continue
2905 applied = _discriminated_union.apply_discriminator(cs.copy(), discriminator, remaining_defs)
2906 # Mutate the schema directly to have the discriminator applied
2907 cs.clear() # pyright: ignore[reportAttributeAccessIssue]
2908 cs.update(applied) # pyright: ignore
2910 if remaining_defs:
2911 schema = core_schema.definitions_schema(schema=schema, definitions=[*remaining_defs.values()])
2912 return schema
2914 def _resolve_definition(self, ref: str, definitions: dict[str, CoreSchema], kept_refs: set[str]) -> CoreSchema:
2915 definition = definitions[ref]
2916 if definition['type'] != 'definition-ref':
2917 return definition
2919 # Some `'definition-ref'` schemas might act as "intermediate" references (e.g. when using
2920 # a PEP 695 type alias (which is referenceable) that references another PEP 695 type alias):
2921 visited: set[str] = set()
2922 while definition['type'] == 'definition-ref' and _inlining_behavior(definition) == 'inline':
2923 schema_ref = definition['schema_ref']
2924 if schema_ref in kept_refs and schema_ref != ref:
2925 # The reference chain leads to a *different* definition that is kept on its own (e.g. a
2926 # recursive model referenced through a PEP 695 type alias). Resolve to a reference
2927 # pointing to it instead of following the chain and inlining a copy of that definition,
2928 # which could otherwise introduce a reference cycle in the resulting schema.
2929 # Note: when `schema_ref == ref`, the reference points back to itself, which is a genuine
2930 # circular reference and is reported as an error by the `visited` logic below.
2931 return {**definition, 'ref': ref}
2932 if schema_ref in visited:
2933 raise PydanticUserError(
2934 f'{ref} contains a circular reference to itself.', code='circular-reference-schema'
2935 )
2936 visited.add(schema_ref)
2937 definition = definitions[schema_ref]
2938 return {**definition, 'ref': ref} # pyright: ignore[reportReturnType]
2941class _FieldNameStack:
2942 __slots__ = ('_stack',)
2944 def __init__(self) -> None:
2945 self._stack: list[str] = []
2947 @contextmanager
2948 def push(self, field_name: str) -> Iterator[None]:
2949 self._stack.append(field_name)
2950 try:
2951 yield
2952 finally:
2953 self._stack.pop()
2955 def get(self) -> str | None:
2956 if self._stack:
2957 return self._stack[-1]
2958 else:
2959 return None
2962class _ModelTypeStack:
2963 __slots__ = ('_stack',)
2965 def __init__(self) -> None:
2966 self._stack: list[type] = []
2968 @contextmanager
2969 def push(self, type_obj: type) -> Iterator[None]:
2970 self._stack.append(type_obj)
2971 try:
2972 yield
2973 finally:
2974 self._stack.pop()
2976 def get(self) -> type | None:
2977 if self._stack:
2978 return self._stack[-1]
2979 else:
2980 return None