Coverage for pydantic/_internal/_std_types_schema.py: 95.26%

1"""Logic for generating pydantic-core schemas for standard library types. 

2 

3Import of this module is deferred since it contains imports of many standard library modules. 

4""" 

5 

6from __future__ import annotations as _annotations 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

7 

8import collections 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

9import collections.abc 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

10import dataclasses 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

11import decimal 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

12import inspect 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

13import os 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

14import typing 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

15from enum import Enum 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

16from functools import partial 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

17from ipaddress import IPv4Address, IPv4Interface, IPv4Network, IPv6Address, IPv6Interface, IPv6Network 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

18from operator import attrgetter 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

19from typing import Any, Callable, Iterable, Literal, Tuple, TypeVar 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

20 

21import typing_extensions 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

22from pydantic_core import ( 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

23 CoreSchema, 

24 MultiHostUrl, 

25 PydanticCustomError, 

26 PydanticOmit, 

27 Url, 

28 core_schema, 

29) 

30from typing_extensions import get_args, get_origin 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

31 

32from pydantic.errors import PydanticSchemaGenerationError 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

33from pydantic.fields import FieldInfo 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

34from pydantic.types import Strict 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

35 

36from ..config import ConfigDict 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

37from ..json_schema import JsonSchemaValue 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

38from . import _known_annotated_metadata, _typing_extra, _validators 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

39from ._core_utils import get_type_ref 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

40from ._internal_dataclass import slots_true 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

41from ._schema_generation_shared import GetCoreSchemaHandler, GetJsonSchemaHandler 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

42 

43if typing.TYPE_CHECKING: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

44 from ._generate_schema import GenerateSchema 

45 

46 StdSchemaFunction = Callable[[GenerateSchema, type[Any]], core_schema.CoreSchema] 

47 

48 

49@dataclasses.dataclass(**slots_true) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

50class SchemaTransformer: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

51 get_core_schema: Callable[[Any, GetCoreSchemaHandler], CoreSchema] 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

52 get_json_schema: Callable[[CoreSchema, GetJsonSchemaHandler], JsonSchemaValue] 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

53 

54 def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> CoreSchema: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

55 return self.get_core_schema(source_type, handler) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

56 

57 def __get_pydantic_json_schema__(self, schema: CoreSchema, handler: GetJsonSchemaHandler) -> JsonSchemaValue: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

58 return self.get_json_schema(schema, handler) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

59 

60 

61def get_enum_core_schema(enum_type: type[Enum], config: ConfigDict) -> CoreSchema: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

62 cases: list[Any] = list(enum_type.__members__.values()) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

63 

64 enum_ref = get_type_ref(enum_type) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

65 description = None if not enum_type.__doc__ else inspect.cleandoc(enum_type.__doc__) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

66 if description == 'An enumeration.': # This is the default value provided by enum.EnumMeta.__new__; don't use it 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

67 description = None 20 ctx1ABabcdefCDghijEFklmn

68 js_updates = {'title': enum_type.__name__, 'description': description} 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

69 js_updates = {k: v for k, v in js_updates.items() if v is not None} 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

70 

71 sub_type: Literal['str', 'int', 'float'] | None = None 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

72 if issubclass(enum_type, int): 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

73 sub_type = 'int' 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

74 value_ser_type: core_schema.SerSchema = core_schema.simple_ser_schema('int') 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

75 elif issubclass(enum_type, str): 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

76 # this handles `StrEnum` (3.11 only), and also `Foobar(str, Enum)` 

77 sub_type = 'str' 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

78 value_ser_type = core_schema.simple_ser_schema('str') 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

79 elif issubclass(enum_type, float): 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

80 sub_type = 'float' 26 ctx1abcdopqrefghijstuvklmnwxyz

81 value_ser_type = core_schema.simple_ser_schema('float') 26 ctx1abcdopqrefghijstuvklmnwxyz

82 else: 

83 # TODO this is an ugly hack, how do we trigger an Any schema for serialization? 

84 value_ser_type = core_schema.plain_serializer_function_ser_schema(lambda x: x) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

85 

86 if cases: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

87 

88 def get_json_schema(schema: CoreSchema, handler: GetJsonSchemaHandler) -> JsonSchemaValue: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

89 json_schema = handler(schema) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

90 original_schema = handler.resolve_ref_schema(json_schema) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

91 original_schema.update(js_updates) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

92 return json_schema 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

93 

94 # we don't want to add the missing to the schema if it's the default one 

95 default_missing = getattr(enum_type._missing_, '__func__', None) == Enum._missing_.__func__ # type: ignore 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

96 enum_schema = core_schema.enum_schema( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

97 enum_type, 

98 cases, 

99 sub_type=sub_type, 

100 missing=None if default_missing else enum_type._missing_, 

101 ref=enum_ref, 

102 metadata={'pydantic_js_functions': [get_json_schema]}, 

103 ) 

104 

105 if config.get('use_enum_values', False): 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

106 enum_schema = core_schema.no_info_after_validator_function( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

107 attrgetter('value'), enum_schema, serialization=value_ser_type 

108 ) 

109 

110 return enum_schema 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

111 

112 else: 

113 

114 def get_json_schema_no_cases(_, handler: GetJsonSchemaHandler) -> JsonSchemaValue: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

115 json_schema = handler(core_schema.enum_schema(enum_type, cases, sub_type=sub_type, ref=enum_ref)) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

116 original_schema = handler.resolve_ref_schema(json_schema) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

117 original_schema.update(js_updates) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

118 return json_schema 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

119 

120 # Use an isinstance check for enums with no cases. 

121 # The most important use case for this is creating TypeVar bounds for generics that should 

122 # be restricted to enums. This is more consistent than it might seem at first, since you can only 

123 # subclass enum.Enum (or subclasses of enum.Enum) if all parent classes have no cases. 

124 # We use the get_json_schema function when an Enum subclass has been declared with no cases 

125 # so that we can still generate a valid json schema. 

126 return core_schema.is_instance_schema( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

127 enum_type, 

128 metadata={'pydantic_js_functions': [get_json_schema_no_cases]}, 

129 ) 

130 

131 

132@dataclasses.dataclass(**slots_true) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

133class InnerSchemaValidator: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

134 """Use a fixed CoreSchema, avoiding interference from outward annotations.""" 

135 

136 core_schema: CoreSchema 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

137 js_schema: JsonSchemaValue | None = None 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

138 js_core_schema: CoreSchema | None = None 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

139 js_schema_update: JsonSchemaValue | None = None 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

140 

141 def __get_pydantic_json_schema__(self, _schema: CoreSchema, handler: GetJsonSchemaHandler) -> JsonSchemaValue: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

142 if self.js_schema is not None: 142 ↛ 143line 142 didn't jump to line 143, because the condition on line 142 was never true32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

143 return self.js_schema 

144 js_schema = handler(self.js_core_schema or self.core_schema) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

145 if self.js_schema_update is not None: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

146 js_schema.update(self.js_schema_update) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

147 return js_schema 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

148 

149 def __get_pydantic_core_schema__(self, _source_type: Any, _handler: GetCoreSchemaHandler) -> CoreSchema: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

150 return self.core_schema 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

151 

152 

153def decimal_prepare_pydantic_annotations( 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

154 source: Any, annotations: Iterable[Any], config: ConfigDict 

155) -> tuple[Any, list[Any]] | None: 

156 if source is not decimal.Decimal: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

157 return None 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

158 

159 metadata, remaining_annotations = _known_annotated_metadata.collect_known_metadata(annotations) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

160 

161 config_allow_inf_nan = config.get('allow_inf_nan') 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

162 if config_allow_inf_nan is not None: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

163 metadata.setdefault('allow_inf_nan', config_allow_inf_nan) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

164 

165 _known_annotated_metadata.check_metadata( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

166 metadata, {*_known_annotated_metadata.FLOAT_CONSTRAINTS, 'max_digits', 'decimal_places'}, decimal.Decimal 

167 ) 

168 return source, [InnerSchemaValidator(core_schema.decimal_schema(**metadata)), *remaining_annotations] 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

169 

170 

171def datetime_prepare_pydantic_annotations( 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

172 source_type: Any, annotations: Iterable[Any], _config: ConfigDict 

173) -> tuple[Any, list[Any]] | None: 

174 import datetime 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

175 

176 metadata, remaining_annotations = _known_annotated_metadata.collect_known_metadata(annotations) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

177 if source_type is datetime.date: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

178 sv = InnerSchemaValidator(core_schema.date_schema(**metadata)) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

179 elif source_type is datetime.datetime: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

180 sv = InnerSchemaValidator(core_schema.datetime_schema(**metadata)) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

181 elif source_type is datetime.time: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

182 sv = InnerSchemaValidator(core_schema.time_schema(**metadata)) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

183 elif source_type is datetime.timedelta: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

184 sv = InnerSchemaValidator(core_schema.timedelta_schema(**metadata)) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

185 else: 

186 return None 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

187 # check now that we know the source type is correct 

188 _known_annotated_metadata.check_metadata(metadata, _known_annotated_metadata.DATE_TIME_CONSTRAINTS, source_type) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

189 return (source_type, [sv, *remaining_annotations]) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

190 

191 

192def uuid_prepare_pydantic_annotations( 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

193 source_type: Any, annotations: Iterable[Any], _config: ConfigDict 

194) -> tuple[Any, list[Any]] | None: 

195 # UUIDs have no constraints - they are fixed length, constructing a UUID instance checks the length 

196 

197 from uuid import UUID 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

198 

199 if source_type is not UUID: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

200 return None 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

201 

202 return (source_type, [InnerSchemaValidator(core_schema.uuid_schema()), *annotations]) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

203 

204 

205def path_schema_prepare_pydantic_annotations( 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

206 source_type: Any, annotations: Iterable[Any], _config: ConfigDict 

207) -> tuple[Any, list[Any]] | None: 

208 import pathlib 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

209 

210 if source_type not in { 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

211 os.PathLike, 

212 pathlib.Path, 

213 pathlib.PurePath, 

214 pathlib.PosixPath, 

215 pathlib.PurePosixPath, 

216 pathlib.PureWindowsPath, 

217 }: 

218 return None 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

219 

220 metadata, remaining_annotations = _known_annotated_metadata.collect_known_metadata(annotations) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

221 _known_annotated_metadata.check_metadata(metadata, _known_annotated_metadata.STR_CONSTRAINTS, source_type) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

222 

223 construct_path = pathlib.PurePath if source_type is os.PathLike else source_type 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

224 

225 def path_validator(input_value: str) -> os.PathLike[Any]: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

226 try: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

227 return construct_path(input_value) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

228 except TypeError as e: 

229 raise PydanticCustomError('path_type', 'Input is not a valid path') from e 

230 

231 constrained_str_schema = core_schema.str_schema(**metadata) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

232 

233 instance_schema = core_schema.json_or_python_schema( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

234 json_schema=core_schema.no_info_after_validator_function(path_validator, constrained_str_schema), 

235 python_schema=core_schema.is_instance_schema(source_type), 

236 ) 

237 

238 strict: bool | None = None 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

239 for annotation in annotations: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

240 if isinstance(annotation, Strict): 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

241 strict = annotation.strict 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

242 

243 schema = core_schema.lax_or_strict_schema( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

244 lax_schema=core_schema.union_schema( 

245 [ 

246 instance_schema, 

247 core_schema.no_info_after_validator_function(path_validator, constrained_str_schema), 

248 ], 

249 custom_error_type='path_type', 

250 custom_error_message='Input is not a valid path', 

251 strict=True, 

252 ), 

253 strict_schema=instance_schema, 

254 serialization=core_schema.to_string_ser_schema(), 

255 strict=strict, 

256 ) 

257 

258 return ( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

259 source_type, 

260 [ 

261 InnerSchemaValidator(schema, js_core_schema=constrained_str_schema, js_schema_update={'format': 'path'}), 

262 *remaining_annotations, 

263 ], 

264 ) 

265 

266 

267def dequeue_validator( 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

268 input_value: Any, handler: core_schema.ValidatorFunctionWrapHandler, maxlen: None | int 

269) -> collections.deque[Any]: 

270 if isinstance(input_value, collections.deque): 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

271 maxlens = [v for v in (input_value.maxlen, maxlen) if v is not None] 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

272 if maxlens: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

273 maxlen = min(maxlens) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

274 return collections.deque(handler(input_value), maxlen=maxlen) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

275 else: 

276 return collections.deque(handler(input_value), maxlen=maxlen) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

277 

278 

279def serialize_sequence_via_list( 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

280 v: Any, handler: core_schema.SerializerFunctionWrapHandler, info: core_schema.SerializationInfo 

281) -> Any: 

282 items: list[Any] = [] 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

283 

284 mapped_origin = SEQUENCE_ORIGIN_MAP.get(type(v), None) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

285 if mapped_origin is None: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

286 # we shouldn't hit this branch, should probably add a serialization error or something 

287 return v 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

288 

289 for index, item in enumerate(v): 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

290 try: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

291 v = handler(item, index) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

292 except PydanticOmit: 

293 pass 

294 else: 

295 items.append(v) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

296 

297 if info.mode_is_json(): 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

298 return items 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

299 else: 

300 return mapped_origin(items) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

301 

302 

303@dataclasses.dataclass(**slots_true) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

304class SequenceValidator: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

305 mapped_origin: type[Any] 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

306 item_source_type: type[Any] 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

307 min_length: int | None = None 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

308 max_length: int | None = None 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

309 strict: bool | None = None 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

310 

311 def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> CoreSchema: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

312 if self.item_source_type is Any: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

313 items_schema = None 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

314 else: 

315 items_schema = handler.generate_schema(self.item_source_type) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

316 

317 metadata = {'min_length': self.min_length, 'max_length': self.max_length, 'strict': self.strict} 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

318 

319 if self.mapped_origin in (list, set, frozenset): 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

320 if self.mapped_origin is list: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

321 constrained_schema = core_schema.list_schema(items_schema, **metadata) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

322 elif self.mapped_origin is set: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

323 constrained_schema = core_schema.set_schema(items_schema, **metadata) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

324 else: 

325 assert self.mapped_origin is frozenset # safety check in case we forget to add a case 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

326 constrained_schema = core_schema.frozenset_schema(items_schema, **metadata) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

327 

328 schema = constrained_schema 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

329 else: 

330 # safety check in case we forget to add a case 

331 assert self.mapped_origin in (collections.deque, collections.Counter) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

332 

333 if self.mapped_origin is collections.deque: 333 ↛ 342line 333 didn't jump to line 342, because the condition on line 333 was always true32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

334 # if we have a MaxLen annotation might as well set that as the default maxlen on the deque 

335 # this lets us re-use existing metadata annotations to let users set the maxlen on a dequeue 

336 # that e.g. comes from JSON 

337 coerce_instance_wrap = partial( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

338 core_schema.no_info_wrap_validator_function, 

339 partial(dequeue_validator, maxlen=metadata.get('max_length', None)), 

340 ) 

341 else: 

342 coerce_instance_wrap = partial(core_schema.no_info_after_validator_function, self.mapped_origin) 

343 

344 # we have to use a lax list schema here, because we need to validate the deque's 

345 # items via a list schema, but it's ok if the deque itself is not a list (same for Counter) 

346 metadata_with_strict_override = {**metadata, 'strict': False} 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

347 constrained_schema = core_schema.list_schema(items_schema, **metadata_with_strict_override) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

348 

349 check_instance = core_schema.json_or_python_schema( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

350 json_schema=core_schema.list_schema(), 

351 python_schema=core_schema.is_instance_schema(self.mapped_origin), 

352 ) 

353 

354 serialization = core_schema.wrap_serializer_function_ser_schema( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

355 serialize_sequence_via_list, schema=items_schema or core_schema.any_schema(), info_arg=True 

356 ) 

357 

358 strict = core_schema.chain_schema([check_instance, coerce_instance_wrap(constrained_schema)]) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

359 

360 if metadata.get('strict', False): 360 ↛ 361line 360 didn't jump to line 361, because the condition on line 360 was never true32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

361 schema = strict 

362 else: 

363 lax = coerce_instance_wrap(constrained_schema) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

364 schema = core_schema.lax_or_strict_schema(lax_schema=lax, strict_schema=strict) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

365 schema['serialization'] = serialization 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

366 

367 return schema 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

368 

369 

370SEQUENCE_ORIGIN_MAP: dict[Any, Any] = { 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

371 typing.Deque: collections.deque, 

372 collections.deque: collections.deque, 

373 list: list, 

374 typing.List: list, 

375 set: set, 

376 typing.AbstractSet: set, 

377 typing.Set: set, 

378 frozenset: frozenset, 

379 typing.FrozenSet: frozenset, 

380 typing.Sequence: list, 

381 typing.MutableSequence: list, 

382 typing.MutableSet: set, 

383 # this doesn't handle subclasses of these 

384 # parametrized typing.Set creates one of these 

385 collections.abc.MutableSet: set, 

386 collections.abc.Set: frozenset, 

387} 

388 

389 

390def identity(s: CoreSchema) -> CoreSchema: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

391 return s 

392 

393 

394def sequence_like_prepare_pydantic_annotations( 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

395 source_type: Any, annotations: Iterable[Any], _config: ConfigDict 

396) -> tuple[Any, list[Any]] | None: 

397 origin: Any = get_origin(source_type) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

398 

399 mapped_origin = SEQUENCE_ORIGIN_MAP.get(origin, None) if origin else SEQUENCE_ORIGIN_MAP.get(source_type, None) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

400 if mapped_origin is None: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

401 return None 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

402 

403 args = get_args(source_type) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

404 

405 if not args: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

406 args = typing.cast(Tuple[Any], (Any,)) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

407 elif len(args) != 1: 407 ↛ 408line 407 didn't jump to line 408, because the condition on line 407 was never true41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

408 raise ValueError('Expected sequence to have exactly 1 generic parameter') 

409 

410 item_source_type = args[0] 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

411 

412 metadata, remaining_annotations = _known_annotated_metadata.collect_known_metadata(annotations) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

413 _known_annotated_metadata.check_metadata(metadata, _known_annotated_metadata.SEQUENCE_CONSTRAINTS, source_type) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

414 

415 return (source_type, [SequenceValidator(mapped_origin, item_source_type, **metadata), *remaining_annotations]) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

416 

417 

418MAPPING_ORIGIN_MAP: dict[Any, Any] = { 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

419 typing.DefaultDict: collections.defaultdict, 

420 collections.defaultdict: collections.defaultdict, 

421 collections.OrderedDict: collections.OrderedDict, 

422 typing_extensions.OrderedDict: collections.OrderedDict, 

423 dict: dict, 

424 typing.Dict: dict, 

425 collections.Counter: collections.Counter, 

426 typing.Counter: collections.Counter, 

427 # this doesn't handle subclasses of these 

428 typing.Mapping: dict, 

429 typing.MutableMapping: dict, 

430 # parametrized typing.{Mutable}Mapping creates one of these 

431 collections.abc.MutableMapping: dict, 

432 collections.abc.Mapping: dict, 

433} 

434 

435 

436def defaultdict_validator( 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

437 input_value: Any, handler: core_schema.ValidatorFunctionWrapHandler, default_default_factory: Callable[[], Any] 

438) -> collections.defaultdict[Any, Any]: 

439 if isinstance(input_value, collections.defaultdict): 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

440 default_factory = input_value.default_factory 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

441 return collections.defaultdict(default_factory, handler(input_value)) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

442 else: 

443 return collections.defaultdict(default_default_factory, handler(input_value)) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

444 

445 

446def get_defaultdict_default_default_factory(values_source_type: Any) -> Callable[[], Any]: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

447 def infer_default() -> Callable[[], Any]: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

448 allowed_default_types: dict[Any, Any] = { 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

449 typing.Tuple: tuple, 

450 tuple: tuple, 

451 collections.abc.Sequence: tuple, 

452 collections.abc.MutableSequence: list, 

453 typing.List: list, 

454 list: list, 

455 typing.Sequence: list, 

456 typing.Set: set, 

457 set: set, 

458 typing.MutableSet: set, 

459 collections.abc.MutableSet: set, 

460 collections.abc.Set: frozenset, 

461 typing.MutableMapping: dict, 

462 typing.Mapping: dict, 

463 collections.abc.Mapping: dict, 

464 collections.abc.MutableMapping: dict, 

465 float: float, 

466 int: int, 

467 str: str, 

468 bool: bool, 

469 } 

470 values_type_origin = get_origin(values_source_type) or values_source_type 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

471 instructions = 'set using `DefaultDict[..., Annotated[..., Field(default_factory=...)]]`' 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

472 if isinstance(values_type_origin, TypeVar): 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

473 

474 def type_var_default_factory() -> None: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

475 raise RuntimeError( 

476 'Generic defaultdict cannot be used without a concrete value type or an' 

477 ' explicit default factory, ' + instructions 

478 ) 

479 

480 return type_var_default_factory 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

481 elif values_type_origin not in allowed_default_types: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

482 # a somewhat subjective set of types that have reasonable default values 

483 allowed_msg = ', '.join([t.__name__ for t in set(allowed_default_types.values())]) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

484 raise PydanticSchemaGenerationError( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

485 f'Unable to infer a default factory for keys of type {values_source_type}.' 

486 f' Only {allowed_msg} are supported, other types require an explicit default factory' 

487 ' ' + instructions 

488 ) 

489 return allowed_default_types[values_type_origin] 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

490 

491 # Assume Annotated[..., Field(...)] 

492 if _typing_extra.is_annotated(values_source_type): 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

493 field_info = next((v for v in get_args(values_source_type) if isinstance(v, FieldInfo)), None) 493 ↛ exitline 493 didn't finish the generator expression on line 49332 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

494 else: 

495 field_info = None 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

496 if field_info and field_info.default_factory: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

497 default_default_factory = field_info.default_factory 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

498 else: 

499 default_default_factory = infer_default() 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

500 return default_default_factory 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

501 

502 

503@dataclasses.dataclass(**slots_true) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

504class MappingValidator: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

505 mapped_origin: type[Any] 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

506 keys_source_type: type[Any] 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

507 values_source_type: type[Any] 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

508 min_length: int | None = None 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

509 max_length: int | None = None 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

510 strict: bool = False 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

511 

512 def serialize_mapping_via_dict(self, v: Any, handler: core_schema.SerializerFunctionWrapHandler) -> Any: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

513 return handler(v) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

514 

515 def __get_pydantic_core_schema__(self, source_type: Any, handler: GetCoreSchemaHandler) -> CoreSchema: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

516 if self.keys_source_type is Any: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

517 keys_schema = None 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

518 else: 

519 keys_schema = handler.generate_schema(self.keys_source_type) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

520 if self.values_source_type is Any: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

521 values_schema = None 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

522 else: 

523 values_schema = handler.generate_schema(self.values_source_type) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

524 

525 metadata = {'min_length': self.min_length, 'max_length': self.max_length, 'strict': self.strict} 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

526 

527 if self.mapped_origin is dict: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

528 schema = core_schema.dict_schema(keys_schema, values_schema, **metadata) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

529 else: 

530 constrained_schema = core_schema.dict_schema(keys_schema, values_schema, **metadata) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

531 check_instance = core_schema.json_or_python_schema( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

532 json_schema=core_schema.dict_schema(), 

533 python_schema=core_schema.is_instance_schema(self.mapped_origin), 

534 ) 

535 

536 if self.mapped_origin is collections.defaultdict: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

537 default_default_factory = get_defaultdict_default_default_factory(self.values_source_type) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

538 coerce_instance_wrap = partial( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

539 core_schema.no_info_wrap_validator_function, 

540 partial(defaultdict_validator, default_default_factory=default_default_factory), 

541 ) 

542 else: 

543 coerce_instance_wrap = partial(core_schema.no_info_after_validator_function, self.mapped_origin) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

544 

545 serialization = core_schema.wrap_serializer_function_ser_schema( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

546 self.serialize_mapping_via_dict, 

547 schema=core_schema.dict_schema( 

548 keys_schema or core_schema.any_schema(), values_schema or core_schema.any_schema() 

549 ), 

550 info_arg=False, 

551 ) 

552 

553 strict = core_schema.chain_schema([check_instance, coerce_instance_wrap(constrained_schema)]) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

554 

555 if metadata.get('strict', False): 555 ↛ 556line 555 didn't jump to line 556, because the condition on line 555 was never true32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

556 schema = strict 

557 else: 

558 lax = coerce_instance_wrap(constrained_schema) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

559 schema = core_schema.lax_or_strict_schema(lax_schema=lax, strict_schema=strict) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

560 schema['serialization'] = serialization 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

561 

562 return schema 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

563 

564 

565def mapping_like_prepare_pydantic_annotations( 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

566 source_type: Any, annotations: Iterable[Any], _config: ConfigDict 

567) -> tuple[Any, list[Any]] | None: 

568 origin: Any = get_origin(source_type) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

569 

570 mapped_origin = MAPPING_ORIGIN_MAP.get(origin, None) if origin else MAPPING_ORIGIN_MAP.get(source_type, None) 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

571 if mapped_origin is None: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

572 return None 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

573 

574 args = get_args(source_type) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

575 

576 if not args: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

577 args = typing.cast(Tuple[Any, Any], (Any, Any)) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

578 elif mapped_origin is collections.Counter: 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

579 # a single generic 

580 if len(args) != 1: 580 ↛ 581line 580 didn't jump to line 581, because the condition on line 580 was never true32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

581 raise ValueError('Expected Counter to have exactly 1 generic parameter') 

582 args = (args[0], int) # keys are always an int 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

583 elif len(args) != 2: 583 ↛ 584line 583 didn't jump to line 584, because the condition on line 583 was never true32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

584 raise ValueError('Expected mapping to have exactly 2 generic parameters') 

585 

586 keys_source_type, values_source_type = args 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

587 

588 metadata, remaining_annotations = _known_annotated_metadata.collect_known_metadata(annotations) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

589 _known_annotated_metadata.check_metadata(metadata, _known_annotated_metadata.SEQUENCE_CONSTRAINTS, source_type) 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

590 

591 return ( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

592 source_type, 

593 [ 

594 MappingValidator(mapped_origin, keys_source_type, values_source_type, **metadata), 

595 *remaining_annotations, 

596 ], 

597 ) 

598 

599 

600def ip_prepare_pydantic_annotations( 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

601 source_type: Any, annotations: Iterable[Any], _config: ConfigDict 

602) -> tuple[Any, list[Any]] | None: 

603 def make_strict_ip_schema(tp: type[Any]) -> CoreSchema: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

604 return core_schema.json_or_python_schema( 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

605 json_schema=core_schema.no_info_after_validator_function(tp, core_schema.str_schema()), 

606 python_schema=core_schema.is_instance_schema(tp), 

607 ) 

608 

609 if source_type is IPv4Address: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

610 return source_type, [ 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

611 SchemaTransformer( 

612 lambda _1, _2: core_schema.lax_or_strict_schema( 

613 lax_schema=core_schema.no_info_plain_validator_function(_validators.ip_v4_address_validator), 

614 strict_schema=make_strict_ip_schema(IPv4Address), 

615 serialization=core_schema.to_string_ser_schema(), 

616 ), 

617 lambda _1, _2: {'type': 'string', 'format': 'ipv4'}, 

618 ), 

619 *annotations, 

620 ] 

621 if source_type is IPv4Network: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

622 return source_type, [ 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

623 SchemaTransformer( 

624 lambda _1, _2: core_schema.lax_or_strict_schema( 

625 lax_schema=core_schema.no_info_plain_validator_function(_validators.ip_v4_network_validator), 

626 strict_schema=make_strict_ip_schema(IPv4Network), 

627 serialization=core_schema.to_string_ser_schema(), 

628 ), 

629 lambda _1, _2: {'type': 'string', 'format': 'ipv4network'}, 

630 ), 

631 *annotations, 

632 ] 

633 if source_type is IPv4Interface: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

634 return source_type, [ 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

635 SchemaTransformer( 

636 lambda _1, _2: core_schema.lax_or_strict_schema( 

637 lax_schema=core_schema.no_info_plain_validator_function(_validators.ip_v4_interface_validator), 

638 strict_schema=make_strict_ip_schema(IPv4Interface), 

639 serialization=core_schema.to_string_ser_schema(), 

640 ), 

641 lambda _1, _2: {'type': 'string', 'format': 'ipv4interface'}, 

642 ), 

643 *annotations, 

644 ] 

645 

646 if source_type is IPv6Address: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

647 return source_type, [ 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

648 SchemaTransformer( 

649 lambda _1, _2: core_schema.lax_or_strict_schema( 

650 lax_schema=core_schema.no_info_plain_validator_function(_validators.ip_v6_address_validator), 

651 strict_schema=make_strict_ip_schema(IPv6Address), 

652 serialization=core_schema.to_string_ser_schema(), 

653 ), 

654 lambda _1, _2: {'type': 'string', 'format': 'ipv6'}, 

655 ), 

656 *annotations, 

657 ] 

658 if source_type is IPv6Network: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

659 return source_type, [ 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

660 SchemaTransformer( 

661 lambda _1, _2: core_schema.lax_or_strict_schema( 

662 lax_schema=core_schema.no_info_plain_validator_function(_validators.ip_v6_network_validator), 

663 strict_schema=make_strict_ip_schema(IPv6Network), 

664 serialization=core_schema.to_string_ser_schema(), 

665 ), 

666 lambda _1, _2: {'type': 'string', 'format': 'ipv6network'}, 

667 ), 

668 *annotations, 

669 ] 

670 if source_type is IPv6Interface: 41 ctx1ABabcdopqrefCDghijstuvGHIJKLMNOEFklmnwxyz

671 return source_type, [ 32 ctx1ABabcdopqrefCDghijstuvEFklmnwxyz

672 SchemaTransformer( 

673 lambda _1, _2: core_schema.lax_or_strict_schema( 

674 lax_schema=core_schema.no_info_plain_validator_function(_validators.ip_v6_interface_validator), 

675 strict_schema=make_strict_ip_schema(IPv6Interface), 

676 serialization=core_schema.to_string_ser_schema(), 

677 ), 

678 lambda _1, _2: {'type': 'string', 'format': 'ipv6interface'}, 

679 ), 

680 *annotations, 

681 ] 

682