Coverage for pydantic/_internal/_generics.py: 93.52%

1from __future__ import annotations 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

2 

3import sys 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

4import types 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

5import typing 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

6from collections import ChainMap 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

7from contextlib import contextmanager 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

8from contextvars import ContextVar 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

9from types import prepare_class 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

10from typing import TYPE_CHECKING, Any, Iterator, List, Mapping, MutableMapping, Tuple, TypeVar 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

11from weakref import WeakValueDictionary 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

12 

13import typing_extensions 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

14 

15from ._core_utils import get_type_ref 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

16from ._forward_ref import PydanticRecursiveRef 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

17from ._typing_extra import TypeVarType, typing_base 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

18from ._utils import all_identical, is_model_class 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

19 

20if sys.version_info >= (3, 10): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

21 from typing import _UnionGenericAlias # type: ignore[attr-defined] 26 ctx1abcdefghijklmGHIJKLMnopqrs

22 

23if TYPE_CHECKING: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

24 from ..main import BaseModel 

25 

26GenericTypesCacheKey = Tuple[Any, Any, Tuple[Any, ...]] 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

27 

28# Note: We want to remove LimitedDict, but to do this, we'd need to improve the handling of generics caching. 

29# Right now, to handle recursive generics, we some types must remain cached for brief periods without references. 

30# By chaining the WeakValuesDict with a LimitedDict, we have a way to retain caching for all types with references, 

31# while also retaining a limited number of types even without references. This is generally enough to build 

32# specific recursive generic models without losing required items out of the cache. 

33 

34KT = TypeVar('KT') 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

35VT = TypeVar('VT') 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

36_LIMITED_DICT_SIZE = 100 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

37if TYPE_CHECKING: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

38 

39 class LimitedDict(dict, MutableMapping[KT, VT]): 

40 def __init__(self, size_limit: int = _LIMITED_DICT_SIZE): ... 

41 

42else: 

43 

44 class LimitedDict(dict): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

45 """Limit the size/length of a dict used for caching to avoid unlimited increase in memory usage. 

46 

47 Since the dict is ordered, and we always remove elements from the beginning, this is effectively a FIFO cache. 

48 """ 

49 

50 def __init__(self, size_limit: int = _LIMITED_DICT_SIZE): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

51 self.size_limit = size_limit 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

52 super().__init__() 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

53 

54 def __setitem__(self, key: Any, value: Any, /) -> None: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

55 super().__setitem__(key, value) 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

56 if len(self) > self.size_limit: 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

57 excess = len(self) - self.size_limit + self.size_limit // 10 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

58 to_remove = list(self.keys())[:excess] 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

59 for k in to_remove: 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

60 del self[k] 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

61 

62 

63# weak dictionaries allow the dynamically created parametrized versions of generic models to get collected 

64# once they are no longer referenced by the caller. 

65if sys.version_info >= (3, 9): # Typing for weak dictionaries available at 3.9 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

66 GenericTypesCache = WeakValueDictionary[GenericTypesCacheKey, 'type[BaseModel]'] 34 ctx1vwabcdefFgzAhijklmOGHIJKLMDEnopqrs

67else: 

68 GenericTypesCache = WeakValueDictionary 7 ctx1tuxyNBC

69 

70if TYPE_CHECKING: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

71 

72 class DeepChainMap(ChainMap[KT, VT]): # type: ignore 

73 ... 

74 

75else: 

76 

77 class DeepChainMap(ChainMap): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

78 """Variant of ChainMap that allows direct updates to inner scopes. 

79 

80 Taken from https://docs.python.org/3/library/collections.html#collections.ChainMap, 

81 with some light modifications for this use case. 

82 """ 

83 

84 def clear(self) -> None: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

85 for mapping in self.maps: 

86 mapping.clear() 

87 

88 def __setitem__(self, key: KT, value: VT) -> None: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

89 for mapping in self.maps: 

90 mapping[key] = value 

91 

92 def __delitem__(self, key: KT) -> None: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

93 hit = False 

94 for mapping in self.maps: 

95 if key in mapping: 

96 del mapping[key] 

97 hit = True 

98 if not hit: 

99 raise KeyError(key) 

100 

101 

102# Despite the fact that LimitedDict _seems_ no longer necessary, I'm very nervous to actually remove it 

103# and discover later on that we need to re-add all this infrastructure... 

104# _GENERIC_TYPES_CACHE = DeepChainMap(GenericTypesCache(), LimitedDict()) 

105 

106_GENERIC_TYPES_CACHE = GenericTypesCache() 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

107 

108 

109class PydanticGenericMetadata(typing_extensions.TypedDict): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

110 origin: type[BaseModel] | None # analogous to typing._GenericAlias.__origin__ 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

111 args: tuple[Any, ...] # analogous to typing._GenericAlias.__args__ 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

112 parameters: tuple[type[Any], ...] # analogous to typing.Generic.__parameters__ 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

113 

114 

115def create_generic_submodel( 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

116 model_name: str, origin: type[BaseModel], args: tuple[Any, ...], params: tuple[Any, ...] 

117) -> type[BaseModel]: 

118 """Dynamically create a submodel of a provided (generic) BaseModel. 

119 

120 This is used when producing concrete parametrizations of generic models. This function 

121 only *creates* the new subclass; the schema/validators/serialization must be updated to 

122 reflect a concrete parametrization elsewhere. 

123 

124 Args: 

125 model_name: The name of the newly created model. 

126 origin: The base class for the new model to inherit from. 

127 args: A tuple of generic metadata arguments. 

128 params: A tuple of generic metadata parameters. 

129 

130 Returns: 

131 The created submodel. 

132 """ 

133 namespace: dict[str, Any] = {'__module__': origin.__module__} 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

134 bases = (origin,) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

135 meta, ns, kwds = prepare_class(model_name, bases) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

136 namespace.update(ns) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

137 created_model = meta( 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

138 model_name, 

139 bases, 

140 namespace, 

141 __pydantic_generic_metadata__={ 

142 'origin': origin, 

143 'args': args, 

144 'parameters': params, 

145 }, 

146 __pydantic_reset_parent_namespace__=False, 

147 **kwds, 

148 ) 

149 

150 model_module, called_globally = _get_caller_frame_info(depth=3) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

151 if called_globally: # create global reference and therefore allow pickling 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

152 object_by_reference = None 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

153 reference_name = model_name 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

154 reference_module_globals = sys.modules[created_model.__module__].__dict__ 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

155 while object_by_reference is not created_model: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

156 object_by_reference = reference_module_globals.setdefault(reference_name, created_model) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

157 reference_name += '_' 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

158 

159 return created_model 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

160 

161 

162def _get_caller_frame_info(depth: int = 2) -> tuple[str | None, bool]: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

163 """Used inside a function to check whether it was called globally. 

164 

165 Args: 

166 depth: The depth to get the frame. 

167 

168 Returns: 

169 A tuple contains `module_name` and `called_globally`. 

170 

171 Raises: 

172 RuntimeError: If the function is not called inside a function. 

173 """ 

174 try: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

175 previous_caller_frame = sys._getframe(depth) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

176 except ValueError as e: 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

177 raise RuntimeError('This function must be used inside another function') from e 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

178 except AttributeError: # sys module does not have _getframe function, so there's nothing we can do about it 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

179 return None, False 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

180 frame_globals = previous_caller_frame.f_globals 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

181 return frame_globals.get('__name__'), previous_caller_frame.f_locals is frame_globals 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

182 

183 

184DictValues: type[Any] = {}.values().__class__ 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

185 

186 

187def iter_contained_typevars(v: Any) -> Iterator[TypeVarType]: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

188 """Recursively iterate through all subtypes and type args of `v` and yield any typevars that are found. 

189 

190 This is inspired as an alternative to directly accessing the `__parameters__` attribute of a GenericAlias, 

191 since __parameters__ of (nested) generic BaseModel subclasses won't show up in that list. 

192 """ 

193 if isinstance(v, TypeVar): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

194 yield v 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

195 elif is_model_class(v): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

196 yield from v.__pydantic_generic_metadata__['parameters'] 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

197 elif isinstance(v, (DictValues, list)): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

198 for var in v: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

199 yield from iter_contained_typevars(var) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

200 else: 

201 args = get_args(v) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

202 for arg in args: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

203 yield from iter_contained_typevars(arg) 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

204 

205 

206def get_args(v: Any) -> Any: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

207 pydantic_generic_metadata: PydanticGenericMetadata | None = getattr(v, '__pydantic_generic_metadata__', None) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

208 if pydantic_generic_metadata: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

209 return pydantic_generic_metadata.get('args') 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

210 return typing_extensions.get_args(v) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

211 

212 

213def get_origin(v: Any) -> Any: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

214 pydantic_generic_metadata: PydanticGenericMetadata | None = getattr(v, '__pydantic_generic_metadata__', None) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

215 if pydantic_generic_metadata: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

216 return pydantic_generic_metadata.get('origin') 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

217 return typing_extensions.get_origin(v) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

218 

219 

220def get_standard_typevars_map(cls: type[Any]) -> dict[TypeVarType, Any] | None: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

221 """Package a generic type's typevars and parametrization (if present) into a dictionary compatible with the 

222 `replace_types` function. Specifically, this works with standard typing generics and typing._GenericAlias. 

223 """ 

224 origin = get_origin(cls) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

225 if origin is None: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

226 return None 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

227 if not hasattr(origin, '__parameters__'): 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

228 return None 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

229 

230 # In this case, we know that cls is a _GenericAlias, and origin is the generic type 

231 # So it is safe to access cls.__args__ and origin.__parameters__ 

232 args: tuple[Any, ...] = cls.__args__ # type: ignore 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

233 parameters: tuple[TypeVarType, ...] = origin.__parameters__ 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

234 return dict(zip(parameters, args)) 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

235 

236 

237def get_model_typevars_map(cls: type[BaseModel]) -> dict[TypeVarType, Any] | None: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

238 """Package a generic BaseModel's typevars and concrete parametrization (if present) into a dictionary compatible 

239 with the `replace_types` function. 

240 

241 Since BaseModel.__class_getitem__ does not produce a typing._GenericAlias, and the BaseModel generic info is 

242 stored in the __pydantic_generic_metadata__ attribute, we need special handling here. 

243 """ 

244 # TODO: This could be unified with `get_standard_typevars_map` if we stored the generic metadata 

245 # in the __origin__, __args__, and __parameters__ attributes of the model. 

246 generic_metadata = cls.__pydantic_generic_metadata__ 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

247 origin = generic_metadata['origin'] 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

248 args = generic_metadata['args'] 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

249 return dict(zip(iter_contained_typevars(origin), args)) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

250 

251 

252def replace_types(type_: Any, type_map: Mapping[Any, Any] | None) -> Any: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

253 """Return type with all occurrences of `type_map` keys recursively replaced with their values. 

254 

255 Args: 

256 type_: The class or generic alias. 

257 type_map: Mapping from `TypeVar` instance to concrete types. 

258 

259 Returns: 

260 A new type representing the basic structure of `type_` with all 

261 `typevar_map` keys recursively replaced. 

262 

263 Example: 

264 ```py 

265 from typing import List, Tuple, Union 

266 

267 from pydantic._internal._generics import replace_types 

268 

269 replace_types(Tuple[str, Union[List[str], float]], {str: int}) 

270 #> Tuple[int, Union[List[int], float]] 

271 ``` 

272 """ 

273 if not type_map: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

274 return type_ 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

275 

276 type_args = get_args(type_) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

277 origin_type = get_origin(type_) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

278 

279 if origin_type is typing_extensions.Annotated: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

280 annotated_type, *annotations = type_args 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

281 annotated = replace_types(annotated_type, type_map) 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

282 for annotation in annotations: 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

283 annotated = typing_extensions.Annotated[annotated, annotation] 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

284 return annotated 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

285 

286 # Having type args is a good indicator that this is a typing module 

287 # class instantiation or a generic alias of some sort. 

288 if type_args: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

289 resolved_type_args = tuple(replace_types(arg, type_map) for arg in type_args) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

290 if all_identical(type_args, resolved_type_args): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

291 # If all arguments are the same, there is no need to modify the 

292 # type or create a new object at all 

293 return type_ 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

294 if ( 14 ctx1tuvwFgxyzABCDE

295 origin_type is not None 

296 and isinstance(type_, typing_base) 

297 and not isinstance(origin_type, typing_base) 

298 and getattr(type_, '_name', None) is not None 

299 ): 

300 # In python < 3.9 generic aliases don't exist so any of these like `list`, 

301 # `type` or `collections.abc.Callable` need to be translated. 

302 # See: https://www.python.org/dev/peps/pep-0585 

303 origin_type = getattr(typing, type_._name) 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

304 assert origin_type is not None 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

305 # PEP-604 syntax (Ex.: list | str) is represented with a types.UnionType object that does not have __getitem__. 

306 # We also cannot use isinstance() since we have to compare types. 

307 if sys.version_info >= (3, 10) and origin_type is types.UnionType: 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

308 return _UnionGenericAlias(origin_type, resolved_type_args) 19 ctx1abcdefghijklmnopqrs

309 # NotRequired[T] and Required[T] don't support tuple type resolved_type_args, hence the condition below 

310 return origin_type[resolved_type_args[0] if len(resolved_type_args) == 1 else resolved_type_args] 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

311 

312 # We handle pydantic generic models separately as they don't have the same 

313 # semantics as "typing" classes or generic aliases 

314 

315 if not origin_type and is_model_class(type_): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

316 parameters = type_.__pydantic_generic_metadata__['parameters'] 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

317 if not parameters: 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

318 return type_ 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

319 resolved_type_args = tuple(replace_types(t, type_map) for t in parameters) 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

320 if all_identical(parameters, resolved_type_args): 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

321 return type_ 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

322 return type_[resolved_type_args] 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

323 

324 # Handle special case for typehints that can have lists as arguments. 

325 # `typing.Callable[[int, str], int]` is an example for this. 

326 if isinstance(type_, (List, list)): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

327 resolved_list = list(replace_types(element, type_map) for element in type_) 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

328 if all_identical(type_, resolved_list): 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

329 return type_ 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

330 return resolved_list 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

331 

332 # If all else fails, we try to resolve the type directly and otherwise just 

333 # return the input with no modifications. 

334 return type_map.get(type_, type_) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

335 

336 

337def has_instance_in_type(type_: Any, isinstance_target: Any) -> bool: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

338 """Checks if the type, or any of its arbitrary nested args, satisfy 

339 `isinstance(<type>, isinstance_target)`. 

340 """ 

341 if isinstance(type_, isinstance_target): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

342 return True 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

343 

344 type_args = get_args(type_) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

345 origin_type = get_origin(type_) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

346 

347 if origin_type is typing_extensions.Annotated: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

348 annotated_type, *annotations = type_args 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

349 return has_instance_in_type(annotated_type, isinstance_target) 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

350 

351 # Having type args is a good indicator that this is a typing module 

352 # class instantiation or a generic alias of some sort. 

353 if any(has_instance_in_type(a, isinstance_target) for a in type_args): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

354 return True 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

355 

356 # Handle special case for typehints that can have lists as arguments. 

357 # `typing.Callable[[int, str], int]` is an example for this. 

358 if isinstance(type_, (List, list)) and not isinstance(type_, typing_extensions.ParamSpec): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

359 if any(has_instance_in_type(element, isinstance_target) for element in type_): 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

360 return True 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

361 

362 return False 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

363 

364 

365def check_parameters_count(cls: type[BaseModel], parameters: tuple[Any, ...]) -> None: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

366 """Check the generic model parameters count is equal. 

367 

368 Args: 

369 cls: The generic model. 

370 parameters: A tuple of passed parameters to the generic model. 

371 

372 Raises: 

373 TypeError: If the passed parameters count is not equal to generic model parameters count. 

374 """ 

375 actual = len(parameters) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

376 expected = len(cls.__pydantic_generic_metadata__['parameters']) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

377 if actual != expected: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

378 description = 'many' if actual > expected else 'few' 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

379 raise TypeError(f'Too {description} parameters for {cls}; actual {actual}, expected {expected}') 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

380 

381 

382_generic_recursion_cache: ContextVar[set[str] | None] = ContextVar('_generic_recursion_cache', default=None) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

383 

384 

385@contextmanager 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

386def generic_recursion_self_type( 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

387 origin: type[BaseModel], args: tuple[Any, ...] 

388) -> Iterator[PydanticRecursiveRef | None]: 

389 """This contextmanager should be placed around the recursive calls used to build a generic type, 

390 and accept as arguments the generic origin type and the type arguments being passed to it. 

391 

392 If the same origin and arguments are observed twice, it implies that a self-reference placeholder 

393 can be used while building the core schema, and will produce a schema_ref that will be valid in the 

394 final parent schema. 

395 """ 

396 previously_seen_type_refs = _generic_recursion_cache.get() 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

397 if previously_seen_type_refs is None: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

398 previously_seen_type_refs = set() 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

399 token = _generic_recursion_cache.set(previously_seen_type_refs) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

400 else: 

401 token = None 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

402 

403 try: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

404 type_ref = get_type_ref(origin, args_override=args) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

405 if type_ref in previously_seen_type_refs: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

406 self_type = PydanticRecursiveRef(type_ref=type_ref) 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

407 yield self_type 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

408 else: 

409 previously_seen_type_refs.add(type_ref) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

410 yield None 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

411 finally: 

412 if token: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

413 _generic_recursion_cache.reset(token) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

414 

415 

416def recursively_defined_type_refs() -> set[str]: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

417 visited = _generic_recursion_cache.get() 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

418 if not visited: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

419 return set() # not in a generic recursion, so there are no types 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

420 

421 return visited.copy() # don't allow modifications 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

422 

423 

424def get_cached_generic_type_early(parent: type[BaseModel], typevar_values: Any) -> type[BaseModel] | None: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

425 """The use of a two-stage cache lookup approach was necessary to have the highest performance possible for 

426 repeated calls to `__class_getitem__` on generic types (which may happen in tighter loops during runtime), 

427 while still ensuring that certain alternative parametrizations ultimately resolve to the same type. 

428 

429 As a concrete example, this approach was necessary to make Model[List[T]][int] equal to Model[List[int]]. 

430 The approach could be modified to not use two different cache keys at different points, but the 

431 _early_cache_key is optimized to be as quick to compute as possible (for repeated-access speed), and the 

432 _late_cache_key is optimized to be as "correct" as possible, so that two types that will ultimately be the 

433 same after resolving the type arguments will always produce cache hits. 

434 

435 If we wanted to move to only using a single cache key per type, we would either need to always use the 

436 slower/more computationally intensive logic associated with _late_cache_key, or would need to accept 

437 that Model[List[T]][int] is a different type than Model[List[T]][int]. Because we rely on subclass relationships 

438 during validation, I think it is worthwhile to ensure that types that are functionally equivalent are actually 

439 equal. 

440 """ 

441 return _GENERIC_TYPES_CACHE.get(_early_cache_key(parent, typevar_values)) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

442 

443 

444def get_cached_generic_type_late( 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

445 parent: type[BaseModel], typevar_values: Any, origin: type[BaseModel], args: tuple[Any, ...] 

446) -> type[BaseModel] | None: 

447 """See the docstring of `get_cached_generic_type_early` for more information about the two-stage cache lookup.""" 

448 cached = _GENERIC_TYPES_CACHE.get(_late_cache_key(origin, args, typevar_values)) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

449 if cached is not None: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

450 set_cached_generic_type(parent, typevar_values, cached, origin, args) 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

451 return cached 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

452 

453 

454def set_cached_generic_type( 39 ctx1tuvwabcdefxyzAhijklmNOGHIJKLMBCDEnopqrs

455 parent: type[BaseModel], 

456 typevar_values: tuple[Any, ...], 

457 type_: type[BaseModel], 

458 origin: type[BaseModel] | None = None, 

459 args: tuple[Any, ...] | None = None, 

460) -> None: 

461 """See the docstring of `get_cached_generic_type_early` for more information about why items are cached with 

462 two different keys. 

463 """ 

464 _GENERIC_TYPES_CACHE[_early_cache_key(parent, typevar_values)] = type_ 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

465 if len(typevar_values) == 1: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

466 _GENERIC_TYPES_CACHE[_early_cache_key(parent, typevar_values[0])] = type_ 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

467 if origin and args: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

468 _GENERIC_TYPES_CACHE[_late_cache_key(origin, args, typevar_values)] = type_ 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

469 

470 

471def _union_orderings_key(typevar_values: Any) -> Any: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

472 """This is intended to help differentiate between Union types with the same arguments in different order. 

473 

474 Thanks to caching internal to the `typing` module, it is not possible to distinguish between 

475 List[Union[int, float]] and List[Union[float, int]] (and similarly for other "parent" origins besides List) 

476 because `typing` considers Union[int, float] to be equal to Union[float, int]. 

477 

478 However, you _can_ distinguish between (top-level) Union[int, float] vs. Union[float, int]. 

479 Because we parse items as the first Union type that is successful, we get slightly more consistent behavior 

480 if we make an effort to distinguish the ordering of items in a union. It would be best if we could _always_ 

481 get the exact-correct order of items in the union, but that would require a change to the `typing` module itself. 

482 (See https://github.com/python/cpython/issues/86483 for reference.) 

483 """ 

484 if isinstance(typevar_values, tuple): 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

485 args_data = [] 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

486 for value in typevar_values: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

487 args_data.append(_union_orderings_key(value)) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

488 return tuple(args_data) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

489 elif typing_extensions.get_origin(typevar_values) is typing.Union: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

490 return get_args(typevar_values) 32 ctx1tuvwabcdefFgxyzAhijklmBCDEnopqrs

491 else: 

492 return () 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

493 

494 

495def _early_cache_key(cls: type[BaseModel], typevar_values: Any) -> GenericTypesCacheKey: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

496 """This is intended for minimal computational overhead during lookups of cached types. 

497 

498 Note that this is overly simplistic, and it's possible that two different cls/typevar_values 

499 inputs would ultimately result in the same type being created in BaseModel.__class_getitem__. 

500 To handle this, we have a fallback _late_cache_key that is checked later if the _early_cache_key 

501 lookup fails, and should result in a cache hit _precisely_ when the inputs to __class_getitem__ 

502 would result in the same type. 

503 """ 

504 return cls, typevar_values, _union_orderings_key(typevar_values) 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

505 

506 

507def _late_cache_key(origin: type[BaseModel], args: tuple[Any, ...], typevar_values: Any) -> GenericTypesCacheKey: 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs

508 """This is intended for use later in the process of creating a new type, when we have more information 

509 about the exact args that will be passed. If it turns out that a different set of inputs to 

510 __class_getitem__ resulted in the same inputs to the generic type creation process, we can still 

511 return the cached type, and update the cache with the _early_cache_key as well. 

512 """ 

513 # The _union_orderings_key is placed at the start here to ensure there cannot be a collision with an 

514 # _early_cache_key, as that function will always produce a BaseModel subclass as the first item in the key, 

515 # whereas this function will always produce a tuple as the first item in the key. 

516 return _union_orderings_key(typevar_values), origin, args 41 ctx1tuvwabcdefFgxyzAhijklmNOGHIJKLMBCDEnopqrs