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
1from __future__ import annotations
3import operator
4import sys
5import types
6import typing
7from collections import ChainMap
8from collections.abc import Generator, Mapping
9from contextvars import ContextVar
10from functools import reduce
11from itertools import zip_longest
12from types import prepare_class
13from typing import TYPE_CHECKING, Annotated, Any, TypedDict, TypeVar, cast
14from weakref import WeakValueDictionary
16import typing_extensions
17from typing_inspection import typing_objects
18from typing_inspection.introspection import is_union_origin
20from . import _type_refs, _typing_extra
21from ._contextlib import contextmanager
22from ._forward_ref import PydanticRecursiveRef
23from ._utils import all_identical, is_model_class
25if TYPE_CHECKING:
26 from ..main import BaseModel
28GenericTypesCacheKey = tuple[Any, Any, tuple[Any, ...]]
30# Note: We want to remove LimitedDict, but to do this, we'd need to improve the handling of generics caching.
31# Right now, to handle recursive generics, we some types must remain cached for brief periods without references.
32# By chaining the WeakValuesDict with a LimitedDict, we have a way to retain caching for all types with references,
33# while also retaining a limited number of types even without references. This is generally enough to build
34# specific recursive generic models without losing required items out of the cache.
36KT = TypeVar('KT')
37VT = TypeVar('VT')
38_LIMITED_DICT_SIZE = 100
41class LimitedDict(dict[KT, VT]):
42 def __init__(self, size_limit: int = _LIMITED_DICT_SIZE) -> None:
43 self.size_limit = size_limit
44 super().__init__()
46 def __setitem__(self, key: KT, value: VT, /) -> None:
47 super().__setitem__(key, value)
48 if len(self) > self.size_limit:
49 excess = len(self) - self.size_limit + self.size_limit // 10
50 to_remove = list(self.keys())[:excess]
51 for k in to_remove:
52 del self[k]
55# weak dictionaries allow the dynamically created parametrized versions of generic models to get collected
56# once they are no longer referenced by the caller.
57GenericTypesCache = WeakValueDictionary[GenericTypesCacheKey, 'type[BaseModel]']
59if TYPE_CHECKING:
61 class DeepChainMap(ChainMap[KT, VT]): # type: ignore
62 ...
64else:
66 class DeepChainMap(ChainMap):
67 """Variant of ChainMap that allows direct updates to inner scopes.
69 Taken from https://docs.python.org/3/library/collections.html#collections.ChainMap,
70 with some light modifications for this use case.
71 """
73 def clear(self) -> None:
74 for mapping in self.maps:
75 mapping.clear()
77 def __setitem__(self, key: KT, value: VT) -> None:
78 for mapping in self.maps:
79 mapping[key] = value
81 def __delitem__(self, key: KT) -> None:
82 hit = False
83 for mapping in self.maps:
84 if key in mapping:
85 del mapping[key]
86 hit = True
87 if not hit:
88 raise KeyError(key)
91# Despite the fact that LimitedDict _seems_ no longer necessary, I'm very nervous to actually remove it
92# and discover later on that we need to re-add all this infrastructure...
93# _GENERIC_TYPES_CACHE = DeepChainMap(GenericTypesCache(), LimitedDict())
95_GENERIC_TYPES_CACHE = GenericTypesCache()
98class PydanticGenericMetadata(TypedDict):
99 origin: type[BaseModel] | None # analogous to typing._GenericAlias.__origin__
100 args: tuple[Any, ...] # analogous to typing._GenericAlias.__args__
101 parameters: tuple[TypeVar, ...] # analogous to typing.Generic.__parameters__
104def create_generic_submodel(
105 model_name: str, origin: type[BaseModel], args: tuple[Any, ...], params: tuple[Any, ...]
106) -> type[BaseModel]:
107 """Dynamically create a submodel of a provided (generic) BaseModel.
109 This is used when producing concrete parametrizations of generic models. This function
110 only *creates* the new subclass; the schema/validators/serialization must be updated to
111 reflect a concrete parametrization elsewhere.
113 Args:
114 model_name: The name of the newly created model.
115 origin: The base class for the new model to inherit from.
116 args: A tuple of generic metadata arguments.
117 params: A tuple of generic metadata parameters.
119 Returns:
120 The created submodel.
121 """
122 model_module, called_globally = _get_caller_frame_info(depth=3)
123 # If parametrized at the module level, the parametrized class is registered in this module (see below).
124 # Its `__module__` must match. We fall back to the module of the origin for backwards compatibility reasons.
125 module = model_module if called_globally and model_module is not None else origin.__module__
126 namespace: dict[str, Any] = {'__module__': module}
127 # As per https://docs.python.org/3/reference/datamodel.html#slots:
128 # "The action of a __slots__ declaration is not limited to the class where it is defined.
129 # __slots__ declared in parents are available in child classes. However, instances of a
130 # child subclass will get a __dict__ and __weakref__ unless the subclass also defines
131 # __slots__".
132 # Because when users parameterize a generic model, we create a subclass of such generic model
133 # (what happens in this function), they can't control the fact that no slots is defined on the
134 # dynamic subclass, and so even if they defined extra __slots__ on the generic class (which results
135 # in the class *not* being weakref-able), the parameterized class *will* be weakref-able.
136 # For this reason (and to make Pydantic generic models behavior closer to generic aliases),
137 # we forward any slots from the origin:
138 if '__slots__' in origin.__dict__:
139 namespace['__slots__'] = origin.__dict__['__slots__']
140 bases = (origin,)
141 meta, ns, kwds = prepare_class(model_name, bases)
142 namespace.update(ns)
143 created_model = meta(
144 model_name,
145 bases,
146 namespace,
147 __pydantic_generic_metadata__={
148 'origin': origin,
149 'args': args,
150 'parameters': params,
151 },
152 __pydantic_reset_parent_namespace__=False,
153 **kwds,
154 )
156 if called_globally: # create global reference and therefore allow pickling
157 object_by_reference = None
158 reference_name = model_name
159 reference_module_globals = sys.modules[module].__dict__
160 while object_by_reference is not created_model:
161 object_by_reference = reference_module_globals.setdefault(reference_name, created_model)
162 reference_name += '_'
164 return created_model
167def _get_caller_frame_info(depth: int = 2) -> tuple[str | None, bool]:
168 """Used inside a function to check whether it was called globally.
170 Args:
171 depth: The depth to get the frame.
173 Returns:
174 A tuple contains `module_name` and `called_globally`.
176 Raises:
177 RuntimeError: If the function is not called inside a function.
178 """
179 try:
180 previous_caller_frame = sys._getframe(depth)
181 except ValueError as e:
182 raise RuntimeError('This function must be used inside another function') from e
183 except AttributeError: # sys module does not have _getframe function, so there's nothing we can do about it
184 return None, False
185 frame_globals = previous_caller_frame.f_globals
186 return frame_globals.get('__name__'), previous_caller_frame.f_locals is frame_globals
189DictValues: type[Any] = {}.values().__class__
192def iter_contained_typevars(v: Any) -> Generator[TypeVar]:
193 """Recursively iterate through all subtypes and type args of `v` and yield any typevars that are found.
195 This is inspired as an alternative to directly accessing the `__parameters__` attribute of a GenericAlias,
196 since __parameters__ of (nested) generic BaseModel subclasses won't show up in that list.
197 """
198 if isinstance(v, TypeVar):
199 yield v
200 elif is_model_class(v):
201 yield from v.__pydantic_generic_metadata__['parameters']
202 elif isinstance(v, (DictValues, list)):
203 for var in v:
204 yield from iter_contained_typevars(var)
205 else:
206 args = get_args(v)
207 for arg in args:
208 yield from iter_contained_typevars(arg)
211def get_args(v: Any) -> Any:
212 pydantic_generic_metadata: PydanticGenericMetadata | None = getattr(v, '__pydantic_generic_metadata__', None)
213 if pydantic_generic_metadata:
214 return pydantic_generic_metadata.get('args')
215 return typing_extensions.get_args(v)
218def get_origin(v: Any) -> Any:
219 pydantic_generic_metadata: PydanticGenericMetadata | None = getattr(v, '__pydantic_generic_metadata__', None)
220 if pydantic_generic_metadata:
221 return pydantic_generic_metadata.get('origin')
222 return typing_extensions.get_origin(v)
225def get_standard_typevars_map(cls: Any) -> dict[TypeVar, Any] | None:
226 """Package a generic type's typevars and parametrization (if present) into a dictionary compatible with the
227 `replace_types` function. Specifically, this works with standard typing generics and typing._GenericAlias.
228 """
229 origin = get_origin(cls)
230 if origin is None:
231 return None
232 if not hasattr(origin, '__parameters__'):
233 return None
235 # In this case, we know that cls is a _GenericAlias, and origin is the generic type
236 # So it is safe to access cls.__args__ and origin.__parameters__
237 args: tuple[Any, ...] = cls.__args__ # type: ignore
238 parameters: tuple[TypeVar, ...] = origin.__parameters__
239 return dict(zip(parameters, args, strict=True))
242def get_model_typevars_map(cls: type[BaseModel]) -> dict[TypeVar, Any]:
243 """Package a generic BaseModel's typevars and concrete parametrization (if present) into a dictionary compatible
244 with the `replace_types` function.
246 Since BaseModel.__class_getitem__ does not produce a typing._GenericAlias, and the BaseModel generic info is
247 stored in the __pydantic_generic_metadata__ attribute, we need special handling here.
248 """
249 # TODO: This could be unified with `get_standard_typevars_map` if we stored the generic metadata
250 # in the __origin__, __args__, and __parameters__ attributes of the model.
251 generic_metadata = cls.__pydantic_generic_metadata__
252 origin = generic_metadata['origin']
253 args = generic_metadata['args']
254 if not args:
255 # No need to go into `iter_contained_typevars`:
256 return {}
257 return dict(zip(iter_contained_typevars(origin), args, strict=True))
260def replace_types(type_: Any, type_map: Mapping[TypeVar, Any] | None) -> Any:
261 """Return type with all occurrences of `type_map` keys recursively replaced with their values.
263 Args:
264 type_: The class or generic alias.
265 type_map: Mapping from `TypeVar` instance to concrete types.
267 Returns:
268 A new type representing the basic structure of `type_` with all
269 `typevar_map` keys recursively replaced.
271 Example:
272 ```python
273 from pydantic._internal._generics import replace_types
275 replace_types(tuple[str, list[str] | float], {str: int})
276 #> tuple[int, list[int] | float]
277 ```
278 """
279 if not type_map:
280 return type_
282 type_args = get_args(type_)
283 origin_type = get_origin(type_)
285 if typing_objects.is_annotated(origin_type):
286 annotated_type, *annotations = type_args
287 annotated_type = replace_types(annotated_type, type_map)
288 # TODO remove parentheses when we drop support for Python 3.10:
289 return Annotated[(annotated_type, *annotations)]
291 # Having type args is a good indicator that this is a typing special form
292 # instance or a generic alias of some sort.
293 if type_args:
294 resolved_type_args = tuple(replace_types(arg, type_map) for arg in type_args)
295 if all_identical(type_args, resolved_type_args):
296 # If all arguments are the same, there is no need to modify the
297 # type or create a new object at all
298 return type_
300 if (
301 origin_type is not None
302 and isinstance(type_, _typing_extra.typing_base)
303 and not isinstance(origin_type, _typing_extra.typing_base)
304 and getattr(type_, '_name', None) is not None
305 ):
306 # In python < 3.9 generic aliases don't exist so any of these like `list`,
307 # `type` or `collections.abc.Callable` need to be translated.
308 # See: https://www.python.org/dev/peps/pep-0585
309 origin_type = getattr(typing, type_._name)
310 assert origin_type is not None
312 if is_union_origin(origin_type):
313 if any(typing_objects.is_any(arg) for arg in resolved_type_args):
314 # `Any | T` ~ `Any`:
315 resolved_type_args = (Any,)
316 # `Never | T` ~ `T`:
317 resolved_type_args = tuple(
318 arg
319 for arg in resolved_type_args
320 if not (typing_objects.is_noreturn(arg) or typing_objects.is_never(arg))
321 )
323 # PEP-604 syntax (e.g. `list | str`) is represented with a types.UnionType object that does not
324 # implement `__getitem__()`. In Python 3.14+, `typing.Union` and `types.UnionType` are the same,
325 # and we instead rely on `typing.Union` as it implicitly converts string annotations to `ForwardRef`
326 # instances (this is to avoid type errors as per https://github.com/python/cpython/pull/105366).
327 if sys.version_info < (3, 14) and origin_type is types.UnionType:
328 return reduce(operator.or_, resolved_type_args)
329 # NotRequired[T] and Required[T] don't support tuple type resolved_type_args, hence the condition below
330 return origin_type[resolved_type_args[0] if len(resolved_type_args) == 1 else resolved_type_args]
332 # We handle pydantic generic models separately as they don't have the same
333 # semantics as "typing" classes or generic aliases
335 if not origin_type and is_model_class(type_):
336 parameters = type_.__pydantic_generic_metadata__['parameters']
337 if not parameters:
338 return type_
339 resolved_type_args = tuple(replace_types(t, type_map) for t in parameters)
340 if all_identical(parameters, resolved_type_args):
341 return type_
342 return type_[resolved_type_args]
344 # Handle special case for typehints that can have lists as arguments.
345 # `typing.Callable[[int, str], int]` is an example for this.
346 if isinstance(type_, list):
347 resolved_list = [replace_types(element, type_map) for element in type_]
348 if all_identical(type_, resolved_list):
349 return type_
350 return resolved_list
352 # If all else fails, we try to resolve the type directly and otherwise just
353 # return the input with no modifications.
354 return type_map.get(type_, type_)
357def map_generic_model_arguments(cls: type[BaseModel], args: tuple[Any, ...]) -> dict[TypeVar, Any]:
358 """Return a mapping between the parameters of a generic model and the provided arguments during parameterization.
360 Raises:
361 TypeError: If the number of arguments does not match the parameters (i.e. if providing too few or too many arguments).
363 Example:
364 ```python {test="skip" lint="skip"}
365 class Model[T, U, V = int](BaseModel): ...
367 map_generic_model_arguments(Model, (str, bytes))
368 #> {T: str, U: bytes, V: int}
370 map_generic_model_arguments(Model, (str,))
371 #> TypeError: Too few arguments for <class '__main__.Model'>; actual 1, expected at least 2
373 map_generic_model_arguments(Model, (str, bytes, int, complex))
374 #> TypeError: Too many arguments for <class '__main__.Model'>; actual 4, expected 3
375 ```
377 Note:
378 This function is analogous to the private `typing._check_generic_specialization` function.
379 """
380 parameters = cls.__pydantic_generic_metadata__['parameters']
381 expected_len = len(parameters)
382 typevars_map: dict[TypeVar, Any] = {}
384 _missing = object()
385 for parameter, argument in zip_longest(parameters, args, fillvalue=_missing):
386 if parameter is _missing:
387 raise TypeError(f'Too many arguments for {cls}; actual {len(args)}, expected {expected_len}')
389 if argument is _missing:
390 param = cast(TypeVar, parameter)
391 try:
392 has_default = param.has_default() # pyright: ignore[reportAttributeAccessIssue]
393 except AttributeError:
394 # Happens if using `typing.TypeVar` (and not `typing_extensions`) on Python < 3.13.
395 has_default = False
396 if has_default:
397 # The default might refer to other type parameters. For an example, see:
398 # https://typing.python.org/en/latest/spec/generics.html#type-parameters-as-parameters-to-generics
399 typevars_map[param] = replace_types(param.__default__, typevars_map) # pyright: ignore[reportAttributeAccessIssue]
400 else:
401 expected_len -= sum(hasattr(p, 'has_default') and p.has_default() for p in parameters) # pyright: ignore[reportAttributeAccessIssue]
402 raise TypeError(f'Too few arguments for {cls}; actual {len(args)}, expected at least {expected_len}')
403 else:
404 param = cast(TypeVar, parameter)
405 typevars_map[param] = argument
407 return typevars_map
410_generic_recursion_cache: ContextVar[set[str] | None] = ContextVar('_generic_recursion_cache', default=None)
413@contextmanager
414def generic_recursion_self_type(
415 origin: type[BaseModel], args: tuple[Any, ...]
416) -> Generator[PydanticRecursiveRef | None]:
417 """This contextmanager should be placed around the recursive calls used to build a generic type,
418 and accept as arguments the generic origin type and the type arguments being passed to it.
420 If the same origin and arguments are observed twice, it implies that a self-reference placeholder
421 can be used while building the core schema, and will produce a schema_ref that will be valid in the
422 final parent schema.
423 """
424 previously_seen_type_refs = _generic_recursion_cache.get()
425 if previously_seen_type_refs is None:
426 previously_seen_type_refs = set()
427 token = _generic_recursion_cache.set(previously_seen_type_refs)
428 else:
429 token = None
431 try:
432 type_ref = _type_refs.model_type_ref(origin, args_override=args)
433 if type_ref in previously_seen_type_refs:
434 self_type = PydanticRecursiveRef(type_ref=type_ref)
435 yield self_type
436 else:
437 previously_seen_type_refs.add(type_ref)
438 yield
439 previously_seen_type_refs.remove(type_ref)
440 finally:
441 if token:
442 _generic_recursion_cache.reset(token)
445def recursively_defined_type_refs() -> set[str]:
446 visited = _generic_recursion_cache.get()
447 if not visited:
448 return set() # not in a generic recursion, so there are no types
450 return visited.copy() # don't allow modifications
453def _generic_cache_get(key: GenericTypesCacheKey) -> type[BaseModel] | None:
454 try:
455 return _GENERIC_TYPES_CACHE.get(key)
456 except TypeError: # unhashable typevar values
457 return None
460def _generic_cache_set(key: GenericTypesCacheKey, value: type[BaseModel]) -> None:
461 try:
462 _GENERIC_TYPES_CACHE[key] = value
463 except TypeError: # unhashable typevar values
464 pass
467def get_cached_generic_type_early(parent: type[BaseModel], typevar_values: Any) -> type[BaseModel] | None:
468 """The use of a two-stage cache lookup approach was necessary to have the highest performance possible for
469 repeated calls to `__class_getitem__` on generic types (which may happen in tighter loops during runtime),
470 while still ensuring that certain alternative parametrizations ultimately resolve to the same type.
472 As a concrete example, this approach was necessary to make Model[List[T]][int] equal to Model[List[int]].
473 The approach could be modified to not use two different cache keys at different points, but the
474 _early_cache_key is optimized to be as quick to compute as possible (for repeated-access speed), and the
475 _late_cache_key is optimized to be as "correct" as possible, so that two types that will ultimately be the
476 same after resolving the type arguments will always produce cache hits.
478 If we wanted to move to only using a single cache key per type, we would either need to always use the
479 slower/more computationally intensive logic associated with _late_cache_key, or would need to accept
480 that Model[List[T]][int] is a different type than Model[List[T]][int]. Because we rely on subclass relationships
481 during validation, I think it is worthwhile to ensure that types that are functionally equivalent are actually
482 equal.
483 """
484 return _generic_cache_get(_early_cache_key(parent, typevar_values))
487def get_cached_generic_type_late(
488 parent: type[BaseModel], typevar_values: Any, origin: type[BaseModel], args: tuple[Any, ...]
489) -> type[BaseModel] | None:
490 """See the docstring of `get_cached_generic_type_early` for more information about the two-stage cache lookup."""
491 cached = _generic_cache_get(_late_cache_key(origin, args, typevar_values))
492 if cached is not None:
493 set_cached_generic_type(parent, typevar_values, cached, origin, args)
494 return cached
497def set_cached_generic_type(
498 parent: type[BaseModel],
499 typevar_values: tuple[Any, ...],
500 type_: type[BaseModel],
501 origin: type[BaseModel] | None = None,
502 args: tuple[Any, ...] | None = None,
503) -> None:
504 """See the docstring of `get_cached_generic_type_early` for more information about why items are cached with
505 two different keys.
506 """
507 _generic_cache_set(_early_cache_key(parent, typevar_values), type_)
508 if len(typevar_values) == 1:
509 _generic_cache_set(_early_cache_key(parent, typevar_values[0]), type_)
510 if origin and args:
511 _generic_cache_set(_late_cache_key(origin, args, typevar_values), type_)
514def _union_orderings_key(typevar_values: Any) -> Any:
515 """This is intended to help differentiate between Union types with the same arguments in different order.
517 Thanks to caching internal to the `typing` module, it is not possible to distinguish between
518 List[Union[int, float]] and List[Union[float, int]] (and similarly for other "parent" origins besides List)
519 because `typing` considers Union[int, float] to be equal to Union[float, int].
521 However, you _can_ distinguish between (top-level) Union[int, float] vs. Union[float, int].
522 Because we parse items as the first Union type that is successful, we get slightly more consistent behavior
523 if we make an effort to distinguish the ordering of items in a union. It would be best if we could _always_
524 get the exact-correct order of items in the union, but that would require a change to the `typing` module itself.
525 (See https://github.com/python/cpython/issues/86483 for reference.)
526 """
527 if isinstance(typevar_values, tuple):
528 return tuple(_union_orderings_key(value) for value in typevar_values)
529 elif typing_objects.is_union(typing_extensions.get_origin(typevar_values)):
530 return get_args(typevar_values)
531 else:
532 return ()
535def _early_cache_key(cls: type[BaseModel], typevar_values: Any) -> GenericTypesCacheKey:
536 """This is intended for minimal computational overhead during lookups of cached types.
538 Note that this is overly simplistic, and it's possible that two different cls/typevar_values
539 inputs would ultimately result in the same type being created in BaseModel.__class_getitem__.
540 To handle this, we have a fallback _late_cache_key that is checked later if the _early_cache_key
541 lookup fails, and should result in a cache hit _precisely_ when the inputs to __class_getitem__
542 would result in the same type.
543 """
544 return cls, typevar_values, _union_orderings_key(typevar_values)
547def _late_cache_key(origin: type[BaseModel], args: tuple[Any, ...], typevar_values: Any) -> GenericTypesCacheKey:
548 """This is intended for use later in the process of creating a new type, when we have more information
549 about the exact args that will be passed. If it turns out that a different set of inputs to
550 __class_getitem__ resulted in the same inputs to the generic type creation process, we can still
551 return the cached type, and update the cache with the _early_cache_key as well.
552 """
553 # The _union_orderings_key is placed at the start here to ensure there cannot be a collision with an
554 # _early_cache_key, as that function will always produce a BaseModel subclass as the first item in the key,
555 # whereas this function will always produce a tuple as the first item in the key.
556 return _union_orderings_key(typevar_values), origin, args