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