Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pydantic/_internal/_mock_val_ser.py: 51%

1from __future__ import annotations 

2 

3from collections.abc import Callable, Iterator, Mapping 

4from typing import TYPE_CHECKING, Any, Generic, Literal, Protocol, TypeVar 

5 

6from pydantic_core import CoreSchema, SchemaSerializer, SchemaValidator 

7 

8from ..errors import PydanticErrorCodes, PydanticUserError 

9from ..plugin._schema_validator import PluggableSchemaValidator 

10 

11if TYPE_CHECKING: 

12 from ..dataclasses import PydanticDataclass 

13 from ..main import BaseModel 

14 from ..type_adapter import TypeAdapter 

15 

16 

17ValSer = TypeVar('ValSer', bound=SchemaValidator | PluggableSchemaValidator | SchemaSerializer) 

18T = TypeVar('T') 

19T_co = TypeVar('T_co', covariant=True) 

20 

21 

22class AttemptRebuild(Protocol[T_co]): 

23 def __call__(self, *, parent_namespace_depth: int = ...) -> T_co | None: ... 

24 

25 

26class MockCoreSchema(Mapping[str, Any]): 

27 """Mocker for `pydantic_core.CoreSchema` which optionally attempts to 

28 rebuild the thing it's mocking when one of its methods is accessed and raises an error if that fails. 

29 """ 

30 

31 __slots__ = '_error_message', '_code', '_attempt_rebuild', '_built_memo' 

32 

33 def __init__( 

34 self, 

35 error_message: str, 

36 *, 

37 code: PydanticErrorCodes, 

38 attempt_rebuild: AttemptRebuild[CoreSchema] | None = None, 

39 ) -> None: 

40 self._error_message = error_message 

41 self._code: PydanticErrorCodes = code 

42 self._attempt_rebuild = attempt_rebuild 

43 self._built_memo: CoreSchema | None = None 

44 

45 def __getitem__(self, key: str) -> Any: 

46 return self._get_built().__getitem__(key) 

47 

48 def __len__(self) -> int: 

49 return self._get_built().__len__() 

50 

51 def __iter__(self) -> Iterator[str]: 

52 return self._get_built().__iter__() 

53 

54 def _get_built(self) -> CoreSchema: 

55 if self._built_memo is not None: 

56 return self._built_memo 

57 

58 if self._attempt_rebuild: 

59 schema = self._attempt_rebuild() 

60 if schema is not None: 

61 self._built_memo = schema 

62 return schema 

63 raise PydanticUserError(self._error_message, code=self._code) 

64 

65 def rebuild(self) -> CoreSchema | None: 

66 self._built_memo = None 

67 if self._attempt_rebuild: 

68 schema = self._attempt_rebuild() 

69 if schema is not None: 

70 return schema 

71 else: 

72 raise PydanticUserError(self._error_message, code=self._code) 

73 return None 

74 

75 

76class MockValSer(Generic[ValSer]): 

77 """Mocker for `pydantic_core.SchemaValidator` or `pydantic_core.SchemaSerializer` which optionally attempts to 

78 rebuild the thing it's mocking when one of its methods is accessed and raises an error if that fails. 

79 """ 

80 

81 __slots__ = '_error_message', '_code', '_val_or_ser', '_attempt_rebuild' 

82 

83 def __init__( 

84 self, 

85 error_message: str, 

86 *, 

87 code: PydanticErrorCodes, 

88 val_or_ser: Literal['validator', 'serializer'], 

89 attempt_rebuild: AttemptRebuild[ValSer] | None = None, 

90 ) -> None: 

91 self._error_message = error_message 

92 self._val_or_ser = SchemaValidator if val_or_ser == 'validator' else SchemaSerializer 

93 self._code: PydanticErrorCodes = code 

94 self._attempt_rebuild = attempt_rebuild 

95 

96 def __getattr__(self, item: str) -> None: 

97 __tracebackhide__ = True 

98 # A slot name only reaches `__getattr__()` if it was never set, i.e. the instance was created without `__init__()` 

99 # (when copying/unpickling). 

100 if item in MockValSer.__slots__: 

101 raise AttributeError(item) 

102 if self._attempt_rebuild: 

103 val_ser = self._attempt_rebuild() 

104 if val_ser is not None: 

105 return getattr(val_ser, item) 

106 

107 # raise an AttributeError if `item` doesn't exist 

108 getattr(self._val_or_ser, item) 

109 raise PydanticUserError(self._error_message, code=self._code) 

110 

111 def rebuild(self, *, parent_namespace_depth: int = 5) -> ValSer | None: 

112 if self._attempt_rebuild: 

113 val_ser = self._attempt_rebuild(parent_namespace_depth=parent_namespace_depth) 

114 if val_ser is not None: 

115 return val_ser 

116 else: 

117 raise PydanticUserError(self._error_message, code=self._code) 

118 return None 

119 

120 

121def set_type_adapter_mocks(adapter: TypeAdapter) -> None: 

122 """Set `core_schema`, `validator` and `serializer` to mock core types on a type adapter instance. 

123 

124 Args: 

125 adapter: The type adapter instance to set the mocks on 

126 """ 

127 type_repr = str(adapter._type) 

128 undefined_type_error_message = ( 

129 f'`TypeAdapter[{type_repr}]` is not fully defined; you should define `{type_repr}` and all referenced types,' 

130 f' then call `.rebuild()` on the instance.' 

131 ) 

132 

133 def attempt_rebuild_fn(attr_fn: Callable[[TypeAdapter], T]) -> AttemptRebuild[T]: 

134 def handler(*, parent_namespace_depth: int = 5) -> T | None: 

135 if adapter.rebuild(raise_errors=False, _parent_namespace_depth=parent_namespace_depth) is not False: 

136 return attr_fn(adapter) 

137 return None 

138 

139 return handler 

140 

141 adapter.core_schema = MockCoreSchema( # pyright: ignore[reportAttributeAccessIssue] 

142 undefined_type_error_message, 

143 code='class-not-fully-defined', 

144 attempt_rebuild=attempt_rebuild_fn(lambda ta: ta.core_schema), 

145 ) 

146 adapter.validator = MockValSer( # pyright: ignore[reportAttributeAccessIssue] 

147 undefined_type_error_message, 

148 code='class-not-fully-defined', 

149 val_or_ser='validator', 

150 attempt_rebuild=attempt_rebuild_fn(lambda ta: ta.validator), 

151 ) 

152 adapter.serializer = MockValSer( # pyright: ignore[reportAttributeAccessIssue] 

153 undefined_type_error_message, 

154 code='class-not-fully-defined', 

155 val_or_ser='serializer', 

156 attempt_rebuild=attempt_rebuild_fn(lambda ta: ta.serializer), 

157 ) 

158 

159 

160def set_model_mocks(cls: type[BaseModel], undefined_name: str | None = None) -> None: 

161 """Set `__pydantic_core_schema__`, `__pydantic_validator__` and `__pydantic_serializer__` to mock core types on a model. 

162 

163 Args: 

164 cls: The model class to set the mocks on 

165 undefined_name: Name of the undefined thing, used in error messages 

166 """ 

167 undefined_name = undefined_name if undefined_name is not None else 'all referenced types' 

168 

169 undefined_type_error_message = ( 

170 f'`{cls.__name__}` is not fully defined; you should define {undefined_name},' 

171 f' then call `{cls.__name__}.model_rebuild()`.' 

172 ) 

173 

174 def attempt_rebuild_fn(attr_fn: Callable[[type[BaseModel]], T]) -> AttemptRebuild[T]: 

175 def handler(*, parent_namespace_depth: int = 5) -> T | None: 

176 if cls.model_rebuild(raise_errors=False, _parent_namespace_depth=parent_namespace_depth) is not False: 

177 return attr_fn(cls) 

178 return None 

179 

180 return handler 

181 

182 cls.__pydantic_core_schema__ = MockCoreSchema( # pyright: ignore[reportAttributeAccessIssue] 

183 undefined_type_error_message, 

184 code='class-not-fully-defined', 

185 attempt_rebuild=attempt_rebuild_fn(lambda c: c.__pydantic_core_schema__), 

186 ) 

187 cls.__pydantic_validator__ = MockValSer( # pyright: ignore[reportAttributeAccessIssue] 

188 undefined_type_error_message, 

189 code='class-not-fully-defined', 

190 val_or_ser='validator', 

191 attempt_rebuild=attempt_rebuild_fn(lambda c: c.__pydantic_validator__), 

192 ) 

193 cls.__pydantic_serializer__ = MockValSer( # pyright: ignore[reportAttributeAccessIssue] 

194 undefined_type_error_message, 

195 code='class-not-fully-defined', 

196 val_or_ser='serializer', 

197 attempt_rebuild=attempt_rebuild_fn(lambda c: c.__pydantic_serializer__), 

198 ) 

199 

200 

201def set_dataclass_mocks(cls: type[PydanticDataclass], undefined_name: str | None = None) -> None: 

202 """Set `__pydantic_validator__` and `__pydantic_serializer__` to `MockValSer`s on a dataclass. 

203 

204 Args: 

205 cls: The model class to set the mocks on 

206 undefined_name: Name of the undefined thing, used in error messages 

207 """ 

208 from ..dataclasses import rebuild_dataclass 

209 

210 undefined_name = undefined_name if undefined_name is not None else 'all referenced types' 

211 

212 undefined_type_error_message = ( 

213 f'`{cls.__name__}` is not fully defined; you should define {undefined_name},' 

214 f' then call `pydantic.dataclasses.rebuild_dataclass({cls.__name__})`.' 

215 ) 

216 

217 def attempt_rebuild_fn(attr_fn: Callable[[type[PydanticDataclass]], T]) -> AttemptRebuild[T]: 

218 def handler(*, parent_namespace_depth: int = 5) -> T | None: 

219 if rebuild_dataclass(cls, raise_errors=False, _parent_namespace_depth=parent_namespace_depth) is not False: 

220 return attr_fn(cls) 

221 return None 

222 

223 return handler 

224 

225 cls.__pydantic_core_schema__ = MockCoreSchema( # pyright: ignore[reportAttributeAccessIssue] 

226 undefined_type_error_message, 

227 code='class-not-fully-defined', 

228 attempt_rebuild=attempt_rebuild_fn(lambda c: c.__pydantic_core_schema__), 

229 ) 

230 cls.__pydantic_validator__ = MockValSer( # pyright: ignore[reportAttributeAccessIssue] 

231 undefined_type_error_message, 

232 code='class-not-fully-defined', 

233 val_or_ser='validator', 

234 attempt_rebuild=attempt_rebuild_fn(lambda c: c.__pydantic_validator__), 

235 ) 

236 cls.__pydantic_serializer__ = MockValSer( # pyright: ignore[reportAttributeAccessIssue] 

237 undefined_type_error_message, 

238 code='class-not-fully-defined', 

239 val_or_ser='serializer', 

240 attempt_rebuild=attempt_rebuild_fn(lambda c: c.__pydantic_serializer__), 

241 )