Coverage for pydantic/networks.py: 92.21%

379 statements  

« prev     ^ index     » next       coverage.py v7.6.12, created at 2025-02-13 19:35 +0000

1"""The networks module contains types for common network-related fields.""" 

2 

3from __future__ import annotations as _annotations 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

4 

5import dataclasses as _dataclasses 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

6import re 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

7from dataclasses import fields 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

8from functools import lru_cache 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

9from importlib.metadata import version 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

10from ipaddress import IPv4Address, IPv4Interface, IPv4Network, IPv6Address, IPv6Interface, IPv6Network 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

11from typing import TYPE_CHECKING, Annotated, Any, ClassVar 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

12 

13from pydantic_core import ( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

14 MultiHostHost, 

15 PydanticCustomError, 

16 PydanticSerializationUnexpectedValue, 

17 SchemaSerializer, 

18 core_schema, 

19) 

20from pydantic_core import MultiHostUrl as _CoreMultiHostUrl 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

21from pydantic_core import Url as _CoreUrl 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

22from typing_extensions import Self, TypeAlias 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

23 

24from pydantic.errors import PydanticUserError 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

25 

26from ._internal import _repr, _schema_generation_shared 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

27from ._migration import getattr_migration 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

28from .annotated_handlers import GetCoreSchemaHandler 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

29from .json_schema import JsonSchemaValue 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

30from .type_adapter import TypeAdapter 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

31 

32if TYPE_CHECKING: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

33 import email_validator 

34 

35 NetworkType: TypeAlias = 'str | bytes | int | tuple[str | bytes | int, str | int]' 1a

36 

37else: 

38 email_validator = None 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

39 

40 

41__all__ = [ 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

42 'AnyUrl', 

43 'AnyHttpUrl', 

44 'FileUrl', 

45 'FtpUrl', 

46 'HttpUrl', 

47 'WebsocketUrl', 

48 'AnyWebsocketUrl', 

49 'UrlConstraints', 

50 'EmailStr', 

51 'NameEmail', 

52 'IPvAnyAddress', 

53 'IPvAnyInterface', 

54 'IPvAnyNetwork', 

55 'PostgresDsn', 

56 'CockroachDsn', 

57 'AmqpDsn', 

58 'RedisDsn', 

59 'MongoDsn', 

60 'KafkaDsn', 

61 'NatsDsn', 

62 'validate_email', 

63 'MySQLDsn', 

64 'MariaDBDsn', 

65 'ClickHouseDsn', 

66 'SnowflakeDsn', 

67] 

68 

69 

70@_dataclasses.dataclass 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

71class UrlConstraints: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

72 """Url constraints. 

73 

74 Attributes: 

75 max_length: The maximum length of the url. Defaults to `None`. 

76 allowed_schemes: The allowed schemes. Defaults to `None`. 

77 host_required: Whether the host is required. Defaults to `None`. 

78 default_host: The default host. Defaults to `None`. 

79 default_port: The default port. Defaults to `None`. 

80 default_path: The default path. Defaults to `None`. 

81 """ 

82 

83 max_length: int | None = None 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

84 allowed_schemes: list[str] | None = None 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

85 host_required: bool | None = None 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

86 default_host: str | None = None 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

87 default_port: int | None = None 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

88 default_path: str | None = None 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

89 

90 def __hash__(self) -> int: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

91 return hash( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

92 ( 

93 self.max_length, 

94 tuple(self.allowed_schemes) if self.allowed_schemes is not None else None, 

95 self.host_required, 

96 self.default_host, 

97 self.default_port, 

98 self.default_path, 

99 ) 

100 ) 

101 

102 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

103 def defined_constraints(self) -> dict[str, Any]: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

104 """Fetch a key / value mapping of constraints to values that are not None. Used for core schema updates.""" 

105 return {field.name: value for field in fields(self) if (value := getattr(self, field.name)) is not None} 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

106 

107 def __get_pydantic_core_schema__(self, source: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

108 schema = handler(source) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

109 

110 # for function-wrap schemas, url constraints is applied to the inner schema 

111 # because when we generate schemas for urls, we wrap a core_schema.url_schema() with a function-wrap schema 

112 # that helps with validation on initialization, see _BaseUrl and _BaseMultiHostUrl below. 

113 schema_to_mutate = schema['schema'] if schema['type'] == 'function-wrap' else schema 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

114 if annotated_type := schema_to_mutate['type'] not in ('url', 'multi-host-url'): 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

115 raise PydanticUserError( 

116 f"'UrlConstraints' cannot annotate '{annotated_type}'.", code='invalid-annotated-type' 

117 ) 

118 for constraint_key, constraint_value in self.defined_constraints.items(): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

119 schema_to_mutate[constraint_key] = constraint_value 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

120 return schema 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

121 

122 

123class _BaseUrl: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

124 _constraints: ClassVar[UrlConstraints] = UrlConstraints() 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

125 _url: _CoreUrl 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

126 

127 def __init__(self, url: str | _CoreUrl | _BaseUrl) -> None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

128 self._url = _build_type_adapter(self.__class__).validate_python(url)._url 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

129 

130 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

131 def scheme(self) -> str: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

132 """The scheme part of the URL. 

133 

134 e.g. `https` in `https://user:pass@host:port/path?query#fragment` 

135 """ 

136 return self._url.scheme 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

137 

138 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

139 def username(self) -> str | None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

140 """The username part of the URL, or `None`. 

141 

142 e.g. `user` in `https://user:pass@host:port/path?query#fragment` 

143 """ 

144 return self._url.username 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

145 

146 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

147 def password(self) -> str | None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

148 """The password part of the URL, or `None`. 

149 

150 e.g. `pass` in `https://user:pass@host:port/path?query#fragment` 

151 """ 

152 return self._url.password 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

153 

154 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

155 def host(self) -> str | None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

156 """The host part of the URL, or `None`. 

157 

158 If the URL must be punycode encoded, this is the encoded host, e.g if the input URL is `https://£££.com`, 

159 `host` will be `xn--9aaa.com` 

160 """ 

161 return self._url.host 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

162 

163 def unicode_host(self) -> str | None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

164 """The host part of the URL as a unicode string, or `None`. 

165 

166 e.g. `host` in `https://user:pass@host:port/path?query#fragment` 

167 

168 If the URL must be punycode encoded, this is the decoded host, e.g if the input URL is `https://£££.com`, 

169 `unicode_host()` will be `£££.com` 

170 """ 

171 return self._url.unicode_host() 

172 

173 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

174 def port(self) -> int | None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

175 """The port part of the URL, or `None`. 

176 

177 e.g. `port` in `https://user:pass@host:port/path?query#fragment` 

178 """ 

179 return self._url.port 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

180 

181 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

182 def path(self) -> str | None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

183 """The path part of the URL, or `None`. 

184 

185 e.g. `/path` in `https://user:pass@host:port/path?query#fragment` 

186 """ 

187 return self._url.path 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

188 

189 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

190 def query(self) -> str | None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

191 """The query part of the URL, or `None`. 

192 

193 e.g. `query` in `https://user:pass@host:port/path?query#fragment` 

194 """ 

195 return self._url.query 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

196 

197 def query_params(self) -> list[tuple[str, str]]: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

198 """The query part of the URL as a list of key-value pairs. 

199 

200 e.g. `[('foo', 'bar')]` in `https://user:pass@host:port/path?foo=bar#fragment` 

201 """ 

202 return self._url.query_params() 

203 

204 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

205 def fragment(self) -> str | None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

206 """The fragment part of the URL, or `None`. 

207 

208 e.g. `fragment` in `https://user:pass@host:port/path?query#fragment` 

209 """ 

210 return self._url.fragment 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

211 

212 def unicode_string(self) -> str: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

213 """The URL as a unicode string, unlike `__str__()` this will not punycode encode the host. 

214 

215 If the URL must be punycode encoded, this is the decoded string, e.g if the input URL is `https://£££.com`, 

216 `unicode_string()` will be `https://£££.com` 

217 """ 

218 return self._url.unicode_string() 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

219 

220 def __str__(self) -> str: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

221 """The URL as a string, this will punycode encode the host if required.""" 

222 return str(self._url) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

223 

224 def __repr__(self) -> str: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

225 return f'{self.__class__.__name__}({str(self._url)!r})' 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

226 

227 def __deepcopy__(self, memo: dict) -> Self: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

228 return self.__class__(self._url) 

229 

230 def __eq__(self, other: Any) -> bool: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

231 return self.__class__ is other.__class__ and self._url == other._url 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

232 

233 def __lt__(self, other: Any) -> bool: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

234 return self.__class__ is other.__class__ and self._url < other._url 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

235 

236 def __gt__(self, other: Any) -> bool: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

237 return self.__class__ is other.__class__ and self._url > other._url 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

238 

239 def __le__(self, other: Any) -> bool: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

240 return self.__class__ is other.__class__ and self._url <= other._url 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

241 

242 def __ge__(self, other: Any) -> bool: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

243 return self.__class__ is other.__class__ and self._url >= other._url 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

244 

245 def __hash__(self) -> int: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

246 return hash(self._url) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

247 

248 def __len__(self) -> int: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

249 return len(str(self._url)) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

250 

251 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

252 def build( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

253 cls, 

254 *, 

255 scheme: str, 

256 username: str | None = None, 

257 password: str | None = None, 

258 host: str, 

259 port: int | None = None, 

260 path: str | None = None, 

261 query: str | None = None, 

262 fragment: str | None = None, 

263 ) -> Self: 

264 """Build a new `Url` instance from its component parts. 

265 

266 Args: 

267 scheme: The scheme part of the URL. 

268 username: The username part of the URL, or omit for no username. 

269 password: The password part of the URL, or omit for no password. 

270 host: The host part of the URL. 

271 port: The port part of the URL, or omit for no port. 

272 path: The path part of the URL, or omit for no path. 

273 query: The query part of the URL, or omit for no query. 

274 fragment: The fragment part of the URL, or omit for no fragment. 

275 

276 Returns: 

277 An instance of URL 

278 """ 

279 return cls( 

280 _CoreUrl.build( 

281 scheme=scheme, 

282 username=username, 

283 password=password, 

284 host=host, 

285 port=port, 

286 path=path, 

287 query=query, 

288 fragment=fragment, 

289 ) 

290 ) 

291 

292 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

293 def serialize_url(cls, url: Any, info: core_schema.SerializationInfo) -> str | Self: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

294 if not isinstance(url, cls): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

295 raise PydanticSerializationUnexpectedValue( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

296 f"Expected `{cls}` but got `{type(url)}` with value `'{url}'` - serialized value may not be as expected." 

297 ) 

298 if info.mode == 'json': 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

299 return str(url) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

300 return url 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

301 

302 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

303 def __get_pydantic_core_schema__( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

304 cls, source: type[_BaseUrl], handler: GetCoreSchemaHandler 

305 ) -> core_schema.CoreSchema: 

306 def wrap_val(v, h): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

307 if isinstance(v, source): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

308 return v 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

309 if isinstance(v, _BaseUrl): 309 ↛ 310line 309 didn't jump to line 310 because the condition on line 309 was never true1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

310 v = str(v) 

311 core_url = h(v) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

312 instance = source.__new__(source) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

313 instance._url = core_url 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

314 return instance 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

315 

316 return core_schema.no_info_wrap_validator_function( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

317 wrap_val, 

318 schema=core_schema.url_schema(**cls._constraints.defined_constraints), 

319 serialization=core_schema.plain_serializer_function_ser_schema( 

320 cls.serialize_url, info_arg=True, when_used='always' 

321 ), 

322 ) 

323 

324 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

325 def __get_pydantic_json_schema__( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

326 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler 

327 ) -> JsonSchemaValue: 

328 # we use the url schema for json schema generation, but we might have to extract it from 

329 # the function-wrap schema we use as a tool for validation on initialization 

330 inner_schema = core_schema['schema'] if core_schema['type'] == 'function-wrap' else core_schema 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

331 return handler(inner_schema) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

332 

333 __pydantic_serializer__ = SchemaSerializer(core_schema.any_schema(serialization=core_schema.to_string_ser_schema())) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

334 

335 

336class _BaseMultiHostUrl: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

337 _constraints: ClassVar[UrlConstraints] = UrlConstraints() 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

338 _url: _CoreMultiHostUrl 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

339 

340 def __init__(self, url: str | _CoreMultiHostUrl | _BaseMultiHostUrl) -> None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

341 self._url = _build_type_adapter(self.__class__).validate_python(url)._url 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

342 

343 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

344 def scheme(self) -> str: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

345 """The scheme part of the URL. 

346 

347 e.g. `https` in `https://foo.com,bar.com/path?query#fragment` 

348 """ 

349 return self._url.scheme 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

350 

351 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

352 def path(self) -> str | None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

353 """The path part of the URL, or `None`. 

354 

355 e.g. `/path` in `https://foo.com,bar.com/path?query#fragment` 

356 """ 

357 return self._url.path 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

358 

359 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

360 def query(self) -> str | None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

361 """The query part of the URL, or `None`. 

362 

363 e.g. `query` in `https://foo.com,bar.com/path?query#fragment` 

364 """ 

365 return self._url.query 

366 

367 def query_params(self) -> list[tuple[str, str]]: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

368 """The query part of the URL as a list of key-value pairs. 

369 

370 e.g. `[('foo', 'bar')]` in `https://foo.com,bar.com/path?query#fragment` 

371 """ 

372 return self._url.query_params() 

373 

374 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

375 def fragment(self) -> str | None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

376 """The fragment part of the URL, or `None`. 

377 

378 e.g. `fragment` in `https://foo.com,bar.com/path?query#fragment` 

379 """ 

380 return self._url.fragment 

381 

382 def hosts(self) -> list[MultiHostHost]: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

383 '''The hosts of the `MultiHostUrl` as [`MultiHostHost`][pydantic_core.MultiHostHost] typed dicts. 

384 

385 ```python 

386 from pydantic_core import MultiHostUrl 

387 

388 mhu = MultiHostUrl('https://foo.com:123,foo:bar@bar.com/path') 

389 print(mhu.hosts()) 

390 """ 

391 [ 

392 {'username': None, 'password': None, 'host': 'foo.com', 'port': 123}, 

393 {'username': 'foo', 'password': 'bar', 'host': 'bar.com', 'port': 443} 

394 ] 

395 ``` 

396 Returns: 

397 A list of dicts, each representing a host. 

398 ''' 

399 return self._url.hosts() 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

400 

401 def unicode_string(self) -> str: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

402 """The URL as a unicode string, unlike `__str__()` this will not punycode encode the hosts.""" 

403 return self._url.unicode_string() 

404 

405 def __str__(self) -> str: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

406 """The URL as a string, this will punycode encode the host if required.""" 

407 return str(self._url) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

408 

409 def __repr__(self) -> str: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

410 return f'{self.__class__.__name__}({str(self._url)!r})' 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

411 

412 def __deepcopy__(self, memo: dict) -> Self: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

413 return self.__class__(self._url) 

414 

415 def __eq__(self, other: Any) -> bool: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

416 return self.__class__ is other.__class__ and self._url == other._url 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

417 

418 def __hash__(self) -> int: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

419 return hash(self._url) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

420 

421 def __len__(self) -> int: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

422 return len(str(self._url)) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

423 

424 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

425 def build( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

426 cls, 

427 *, 

428 scheme: str, 

429 hosts: list[MultiHostHost] | None = None, 

430 username: str | None = None, 

431 password: str | None = None, 

432 host: str | None = None, 

433 port: int | None = None, 

434 path: str | None = None, 

435 query: str | None = None, 

436 fragment: str | None = None, 

437 ) -> Self: 

438 """Build a new `MultiHostUrl` instance from its component parts. 

439 

440 This method takes either `hosts` - a list of `MultiHostHost` typed dicts, or the individual components 

441 `username`, `password`, `host` and `port`. 

442 

443 Args: 

444 scheme: The scheme part of the URL. 

445 hosts: Multiple hosts to build the URL from. 

446 username: The username part of the URL. 

447 password: The password part of the URL. 

448 host: The host part of the URL. 

449 port: The port part of the URL. 

450 path: The path part of the URL. 

451 query: The query part of the URL, or omit for no query. 

452 fragment: The fragment part of the URL, or omit for no fragment. 

453 

454 Returns: 

455 An instance of `MultiHostUrl` 

456 """ 

457 return cls( 

458 _CoreMultiHostUrl.build( 

459 scheme=scheme, 

460 hosts=hosts, 

461 username=username, 

462 password=password, 

463 host=host, 

464 port=port, 

465 path=path, 

466 query=query, 

467 fragment=fragment, 

468 ) 

469 ) 

470 

471 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

472 def serialize_url(cls, url: Any, info: core_schema.SerializationInfo) -> str | Self: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

473 if not isinstance(url, cls): 

474 raise PydanticSerializationUnexpectedValue( 

475 f"Expected `{cls}` but got `{type(url)}` with value `'{url}'` - serialized value may not be as expected." 

476 ) 

477 if info.mode == 'json': 

478 return str(url) 

479 return url 

480 

481 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

482 def __get_pydantic_core_schema__( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

483 cls, source: type[_BaseMultiHostUrl], handler: GetCoreSchemaHandler 

484 ) -> core_schema.CoreSchema: 

485 def wrap_val(v, h): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

486 if isinstance(v, source): 486 ↛ 487line 486 didn't jump to line 487 because the condition on line 486 was never true1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

487 return v 

488 if isinstance(v, _BaseMultiHostUrl): 488 ↛ 489line 488 didn't jump to line 489 because the condition on line 488 was never true1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

489 v = str(v) 

490 core_url = h(v) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

491 instance = source.__new__(source) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

492 instance._url = core_url 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

493 return instance 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

494 

495 return core_schema.no_info_wrap_validator_function( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

496 wrap_val, 

497 schema=core_schema.multi_host_url_schema(**cls._constraints.defined_constraints), 

498 serialization=core_schema.plain_serializer_function_ser_schema( 

499 cls.serialize_url, info_arg=True, when_used='always' 

500 ), 

501 ) 

502 

503 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

504 def __get_pydantic_json_schema__( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

505 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler 

506 ) -> JsonSchemaValue: 

507 # we use the url schema for json schema generation, but we might have to extract it from 

508 # the function-wrap schema we use as a tool for validation on initialization 

509 inner_schema = core_schema['schema'] if core_schema['type'] == 'function-wrap' else core_schema 

510 return handler(inner_schema) 

511 

512 __pydantic_serializer__ = SchemaSerializer(core_schema.any_schema(serialization=core_schema.to_string_ser_schema())) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

513 

514 

515@lru_cache 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

516def _build_type_adapter(cls: type[_BaseUrl | _BaseMultiHostUrl]) -> TypeAdapter: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

517 return TypeAdapter(cls) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

518 

519 

520class AnyUrl(_BaseUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

521 """Base type for all URLs. 

522 

523 * Any scheme allowed 

524 * Top-level domain (TLD) not required 

525 * Host not required 

526 

527 Assuming an input URL of `http://samuel:pass@example.com:8000/the/path/?query=here#fragment=is;this=bit`, 

528 the types export the following properties: 

529 

530 - `scheme`: the URL scheme (`http`), always set. 

531 - `host`: the URL host (`example.com`). 

532 - `username`: optional username if included (`samuel`). 

533 - `password`: optional password if included (`pass`). 

534 - `port`: optional port (`8000`). 

535 - `path`: optional path (`/the/path/`). 

536 - `query`: optional URL query (for example, `GET` arguments or "search string", such as `query=here`). 

537 - `fragment`: optional fragment (`fragment=is;this=bit`). 

538 """ 

539 

540 

541# Note: all single host urls inherit from `AnyUrl` to preserve compatibility with pre-v2.10 code 

542# Where urls were annotated variants of `AnyUrl`, which was an alias to `pydantic_core.Url` 

543 

544 

545class AnyHttpUrl(AnyUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

546 """A type that will accept any http or https URL. 

547 

548 * TLD not required 

549 * Host not required 

550 """ 

551 

552 _constraints = UrlConstraints(allowed_schemes=['http', 'https']) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

553 

554 

555class HttpUrl(AnyUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

556 """A type that will accept any http or https URL. 

557 

558 * TLD not required 

559 * Host not required 

560 * Max length 2083 

561 

562 ```python 

563 from pydantic import BaseModel, HttpUrl, ValidationError 

564 

565 class MyModel(BaseModel): 

566 url: HttpUrl 

567 

568 m = MyModel(url='http://www.example.com') # (1)! 

569 print(m.url) 

570 #> http://www.example.com/ 

571 

572 try: 

573 MyModel(url='ftp://invalid.url') 

574 except ValidationError as e: 

575 print(e) 

576 ''' 

577 1 validation error for MyModel 

578 url 

579 URL scheme should be 'http' or 'https' [type=url_scheme, input_value='ftp://invalid.url', input_type=str] 

580 ''' 

581 

582 try: 

583 MyModel(url='not a url') 

584 except ValidationError as e: 

585 print(e) 

586 ''' 

587 1 validation error for MyModel 

588 url 

589 Input should be a valid URL, relative URL without a base [type=url_parsing, input_value='not a url', input_type=str] 

590 ''' 

591 ``` 

592 

593 1. Note: mypy would prefer `m = MyModel(url=HttpUrl('http://www.example.com'))`, but Pydantic will convert the string to an HttpUrl instance anyway. 

594 

595 "International domains" (e.g. a URL where the host or TLD includes non-ascii characters) will be encoded via 

596 [punycode](https://en.wikipedia.org/wiki/Punycode) (see 

597 [this article](https://www.xudongz.com/blog/2017/idn-phishing/) for a good description of why this is important): 

598 

599 ```python 

600 from pydantic import BaseModel, HttpUrl 

601 

602 class MyModel(BaseModel): 

603 url: HttpUrl 

604 

605 m1 = MyModel(url='http://puny£code.com') 

606 print(m1.url) 

607 #> http://xn--punycode-eja.com/ 

608 m2 = MyModel(url='https://www.аррӏе.com/') 

609 print(m2.url) 

610 #> https://www.xn--80ak6aa92e.com/ 

611 m3 = MyModel(url='https://www.example.珠宝/') 

612 print(m3.url) 

613 #> https://www.example.xn--pbt977c/ 

614 ``` 

615 

616 

617 !!! warning "Underscores in Hostnames" 

618 In Pydantic, underscores are allowed in all parts of a domain except the TLD. 

619 Technically this might be wrong - in theory the hostname cannot have underscores, but subdomains can. 

620 

621 To explain this; consider the following two cases: 

622 

623 - `exam_ple.co.uk`: the hostname is `exam_ple`, which should not be allowed since it contains an underscore. 

624 - `foo_bar.example.com` the hostname is `example`, which should be allowed since the underscore is in the subdomain. 

625 

626 Without having an exhaustive list of TLDs, it would be impossible to differentiate between these two. Therefore 

627 underscores are allowed, but you can always do further validation in a validator if desired. 

628 

629 Also, Chrome, Firefox, and Safari all currently accept `http://exam_ple.com` as a URL, so we're in good 

630 (or at least big) company. 

631 """ 

632 

633 _constraints = UrlConstraints(max_length=2083, allowed_schemes=['http', 'https']) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

634 

635 

636class AnyWebsocketUrl(AnyUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

637 """A type that will accept any ws or wss URL. 

638 

639 * TLD not required 

640 * Host not required 

641 """ 

642 

643 _constraints = UrlConstraints(allowed_schemes=['ws', 'wss']) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

644 

645 

646class WebsocketUrl(AnyUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

647 """A type that will accept any ws or wss URL. 

648 

649 * TLD not required 

650 * Host not required 

651 * Max length 2083 

652 """ 

653 

654 _constraints = UrlConstraints(max_length=2083, allowed_schemes=['ws', 'wss']) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

655 

656 

657class FileUrl(AnyUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

658 """A type that will accept any file URL. 

659 

660 * Host not required 

661 """ 

662 

663 _constraints = UrlConstraints(allowed_schemes=['file']) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

664 

665 

666class FtpUrl(AnyUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

667 """A type that will accept ftp URL. 

668 

669 * TLD not required 

670 * Host not required 

671 """ 

672 

673 _constraints = UrlConstraints(allowed_schemes=['ftp']) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

674 

675 

676class PostgresDsn(_BaseMultiHostUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

677 """A type that will accept any Postgres DSN. 

678 

679 * User info required 

680 * TLD not required 

681 * Host required 

682 * Supports multiple hosts 

683 

684 If further validation is required, these properties can be used by validators to enforce specific behaviour: 

685 

686 ```python 

687 from pydantic import ( 

688 BaseModel, 

689 HttpUrl, 

690 PostgresDsn, 

691 ValidationError, 

692 field_validator, 

693 ) 

694 

695 class MyModel(BaseModel): 

696 url: HttpUrl 

697 

698 m = MyModel(url='http://www.example.com') 

699 

700 # the repr() method for a url will display all properties of the url 

701 print(repr(m.url)) 

702 #> HttpUrl('http://www.example.com/') 

703 print(m.url.scheme) 

704 #> http 

705 print(m.url.host) 

706 #> www.example.com 

707 print(m.url.port) 

708 #> 80 

709 

710 class MyDatabaseModel(BaseModel): 

711 db: PostgresDsn 

712 

713 @field_validator('db') 

714 def check_db_name(cls, v): 

715 assert v.path and len(v.path) > 1, 'database must be provided' 

716 return v 

717 

718 m = MyDatabaseModel(db='postgres://user:pass@localhost:5432/foobar') 

719 print(m.db) 

720 #> postgres://user:pass@localhost:5432/foobar 

721 

722 try: 

723 MyDatabaseModel(db='postgres://user:pass@localhost:5432') 

724 except ValidationError as e: 

725 print(e) 

726 ''' 

727 1 validation error for MyDatabaseModel 

728 db 

729 Assertion failed, database must be provided 

730 assert (None) 

731 + where None = PostgresDsn('postgres://user:pass@localhost:5432').path [type=assertion_error, input_value='postgres://user:pass@localhost:5432', input_type=str] 

732 ''' 

733 ``` 

734 """ 

735 

736 _constraints = UrlConstraints( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

737 host_required=True, 

738 allowed_schemes=[ 

739 'postgres', 

740 'postgresql', 

741 'postgresql+asyncpg', 

742 'postgresql+pg8000', 

743 'postgresql+psycopg', 

744 'postgresql+psycopg2', 

745 'postgresql+psycopg2cffi', 

746 'postgresql+py-postgresql', 

747 'postgresql+pygresql', 

748 ], 

749 ) 

750 

751 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

752 def host(self) -> str: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

753 """The required URL host.""" 

754 return self._url.host # pyright: ignore[reportAttributeAccessIssue] 

755 

756 

757class CockroachDsn(AnyUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

758 """A type that will accept any Cockroach DSN. 

759 

760 * User info required 

761 * TLD not required 

762 * Host required 

763 """ 

764 

765 _constraints = UrlConstraints( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

766 host_required=True, 

767 allowed_schemes=[ 

768 'cockroachdb', 

769 'cockroachdb+psycopg2', 

770 'cockroachdb+asyncpg', 

771 ], 

772 ) 

773 

774 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

775 def host(self) -> str: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

776 """The required URL host.""" 

777 return self._url.host # pyright: ignore[reportReturnType] 

778 

779 

780class AmqpDsn(AnyUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

781 """A type that will accept any AMQP DSN. 

782 

783 * User info required 

784 * TLD not required 

785 * Host not required 

786 """ 

787 

788 _constraints = UrlConstraints(allowed_schemes=['amqp', 'amqps']) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

789 

790 

791class RedisDsn(AnyUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

792 """A type that will accept any Redis DSN. 

793 

794 * User info required 

795 * TLD not required 

796 * Host required (e.g., `rediss://:pass@localhost`) 

797 """ 

798 

799 _constraints = UrlConstraints( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

800 allowed_schemes=['redis', 'rediss'], 

801 default_host='localhost', 

802 default_port=6379, 

803 default_path='/0', 

804 host_required=True, 

805 ) 

806 

807 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

808 def host(self) -> str: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

809 """The required URL host.""" 

810 return self._url.host # pyright: ignore[reportReturnType] 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

811 

812 

813class MongoDsn(_BaseMultiHostUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

814 """A type that will accept any MongoDB DSN. 

815 

816 * User info not required 

817 * Database name not required 

818 * Port not required 

819 * User info may be passed without user part (e.g., `mongodb://mongodb0.example.com:27017`). 

820 """ 

821 

822 _constraints = UrlConstraints(allowed_schemes=['mongodb', 'mongodb+srv'], default_port=27017) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

823 

824 

825class KafkaDsn(AnyUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

826 """A type that will accept any Kafka DSN. 

827 

828 * User info required 

829 * TLD not required 

830 * Host not required 

831 """ 

832 

833 _constraints = UrlConstraints(allowed_schemes=['kafka'], default_host='localhost', default_port=9092) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

834 

835 

836class NatsDsn(_BaseMultiHostUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

837 """A type that will accept any NATS DSN. 

838 

839 NATS is a connective technology built for the ever increasingly hyper-connected world. 

840 It is a single technology that enables applications to securely communicate across 

841 any combination of cloud vendors, on-premise, edge, web and mobile, and devices. 

842 More: https://nats.io 

843 """ 

844 

845 _constraints = UrlConstraints( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

846 allowed_schemes=['nats', 'tls', 'ws', 'wss'], default_host='localhost', default_port=4222 

847 ) 

848 

849 

850class MySQLDsn(AnyUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

851 """A type that will accept any MySQL DSN. 

852 

853 * User info required 

854 * TLD not required 

855 * Host not required 

856 """ 

857 

858 _constraints = UrlConstraints( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

859 allowed_schemes=[ 

860 'mysql', 

861 'mysql+mysqlconnector', 

862 'mysql+aiomysql', 

863 'mysql+asyncmy', 

864 'mysql+mysqldb', 

865 'mysql+pymysql', 

866 'mysql+cymysql', 

867 'mysql+pyodbc', 

868 ], 

869 default_port=3306, 

870 host_required=True, 

871 ) 

872 

873 

874class MariaDBDsn(AnyUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

875 """A type that will accept any MariaDB DSN. 

876 

877 * User info required 

878 * TLD not required 

879 * Host not required 

880 """ 

881 

882 _constraints = UrlConstraints( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

883 allowed_schemes=['mariadb', 'mariadb+mariadbconnector', 'mariadb+pymysql'], 

884 default_port=3306, 

885 ) 

886 

887 

888class ClickHouseDsn(AnyUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

889 """A type that will accept any ClickHouse DSN. 

890 

891 * User info required 

892 * TLD not required 

893 * Host not required 

894 """ 

895 

896 _constraints = UrlConstraints( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

897 allowed_schemes=[ 

898 'clickhouse+native', 

899 'clickhouse+asynch', 

900 'clickhouse+http', 

901 'clickhouse', 

902 'clickhouses', 

903 'clickhousedb', 

904 ], 

905 default_host='localhost', 

906 default_port=9000, 

907 ) 

908 

909 

910class SnowflakeDsn(AnyUrl): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

911 """A type that will accept any Snowflake DSN. 

912 

913 * User info required 

914 * TLD not required 

915 * Host required 

916 """ 

917 

918 _constraints = UrlConstraints( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

919 allowed_schemes=['snowflake'], 

920 host_required=True, 

921 ) 

922 

923 @property 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

924 def host(self) -> str: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

925 """The required URL host.""" 

926 return self._url.host # pyright: ignore[reportReturnType] 

927 

928 

929def import_email_validator() -> None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

930 global email_validator 

931 try: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

932 import email_validator 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

933 except ImportError as e: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

934 raise ImportError('email-validator is not installed, run `pip install pydantic[email]`') from e 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

935 if not version('email-validator').partition('.')[0] == '2': 1nbcdeoapfghiqjklm

936 raise ImportError('email-validator version >= 2.0 required, run pip install -U email-validator') 1nbcdeoapfghiqjklm

937 

938 

939if TYPE_CHECKING: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

940 EmailStr = Annotated[str, ...] 

941else: 

942 

943 class EmailStr: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

944 """ 

945 Info: 

946 To use this type, you need to install the optional 

947 [`email-validator`](https://github.com/JoshData/python-email-validator) package: 

948 

949 ```bash 

950 pip install email-validator 

951 ``` 

952 

953 Validate email addresses. 

954 

955 ```python 

956 from pydantic import BaseModel, EmailStr 

957 

958 class Model(BaseModel): 

959 email: EmailStr 

960 

961 print(Model(email='contact@mail.com')) 

962 #> email='contact@mail.com' 

963 ``` 

964 """ # noqa: D212 

965 

966 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

967 def __get_pydantic_core_schema__( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

968 cls, 

969 _source: type[Any], 

970 _handler: GetCoreSchemaHandler, 

971 ) -> core_schema.CoreSchema: 

972 import_email_validator() 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

973 return core_schema.no_info_after_validator_function(cls._validate, core_schema.str_schema()) 1nbcdeoapfghiqjklm

974 

975 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

976 def __get_pydantic_json_schema__( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

977 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler 

978 ) -> JsonSchemaValue: 

979 field_schema = handler(core_schema) 1nbcdeoapfghiqjklm

980 field_schema.update(type='string', format='email') 1nbcdeoapfghiqjklm

981 return field_schema 1nbcdeoapfghiqjklm

982 

983 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

984 def _validate(cls, input_value: str, /) -> str: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

985 return validate_email(input_value)[1] 1nbcdeoapfghiqjklm

986 

987 

988class NameEmail(_repr.Representation): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

989 """ 

990 Info: 

991 To use this type, you need to install the optional 

992 [`email-validator`](https://github.com/JoshData/python-email-validator) package: 

993 

994 ```bash 

995 pip install email-validator 

996 ``` 

997 

998 Validate a name and email address combination, as specified by 

999 [RFC 5322](https://datatracker.ietf.org/doc/html/rfc5322#section-3.4). 

1000 

1001 The `NameEmail` has two properties: `name` and `email`. 

1002 In case the `name` is not provided, it's inferred from the email address. 

1003 

1004 ```python 

1005 from pydantic import BaseModel, NameEmail 

1006 

1007 class User(BaseModel): 

1008 email: NameEmail 

1009 

1010 user = User(email='Fred Bloggs <fred.bloggs@example.com>') 

1011 print(user.email) 

1012 #> Fred Bloggs <fred.bloggs@example.com> 

1013 print(user.email.name) 

1014 #> Fred Bloggs 

1015 

1016 user = User(email='fred.bloggs@example.com') 

1017 print(user.email) 

1018 #> fred.bloggs <fred.bloggs@example.com> 

1019 print(user.email.name) 

1020 #> fred.bloggs 

1021 ``` 

1022 """ # noqa: D212 

1023 

1024 __slots__ = 'name', 'email' 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1025 

1026 def __init__(self, name: str, email: str): 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1027 self.name = name 1nbcdeoapfghiqjklm

1028 self.email = email 1nbcdeoapfghiqjklm

1029 

1030 def __eq__(self, other: Any) -> bool: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1031 return isinstance(other, NameEmail) and (self.name, self.email) == (other.name, other.email) 1nbcdeoapfghiqjklm

1032 

1033 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1034 def __get_pydantic_json_schema__( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1035 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler 

1036 ) -> JsonSchemaValue: 

1037 field_schema = handler(core_schema) 1nbcdeoapfghiqjklm

1038 field_schema.update(type='string', format='name-email') 1nbcdeoapfghiqjklm

1039 return field_schema 1nbcdeoapfghiqjklm

1040 

1041 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1042 def __get_pydantic_core_schema__( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1043 cls, 

1044 _source: type[Any], 

1045 _handler: GetCoreSchemaHandler, 

1046 ) -> core_schema.CoreSchema: 

1047 import_email_validator() 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1048 

1049 return core_schema.no_info_after_validator_function( 1nbcdeoapfghiqjklm

1050 cls._validate, 

1051 core_schema.json_or_python_schema( 

1052 json_schema=core_schema.str_schema(), 

1053 python_schema=core_schema.union_schema( 

1054 [core_schema.is_instance_schema(cls), core_schema.str_schema()], 

1055 custom_error_type='name_email_type', 

1056 custom_error_message='Input is not a valid NameEmail', 

1057 ), 

1058 serialization=core_schema.to_string_ser_schema(), 

1059 ), 

1060 ) 

1061 

1062 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1063 def _validate(cls, input_value: Self | str, /) -> Self: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1064 if isinstance(input_value, str): 1nbcdeoapfghiqjklm

1065 name, email = validate_email(input_value) 1nbcdeoapfghiqjklm

1066 return cls(name, email) 1nbcdeoapfghiqjklm

1067 else: 

1068 return input_value 1nbcdeoapfghiqjklm

1069 

1070 def __str__(self) -> str: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1071 if '@' in self.name: 1nbcdeoapfghiqjklm

1072 return f'"{self.name}" <{self.email}>' 1nbcdeoapfghiqjklm

1073 

1074 return f'{self.name} <{self.email}>' 1nbcdeoapfghiqjklm

1075 

1076 

1077IPvAnyAddressType: TypeAlias = 'IPv4Address | IPv6Address' 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1078IPvAnyInterfaceType: TypeAlias = 'IPv4Interface | IPv6Interface' 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1079IPvAnyNetworkType: TypeAlias = 'IPv4Network | IPv6Network' 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1080 

1081if TYPE_CHECKING: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1082 IPvAnyAddress = IPvAnyAddressType 

1083 IPvAnyInterface = IPvAnyInterfaceType 

1084 IPvAnyNetwork = IPvAnyNetworkType 

1085else: 

1086 

1087 class IPvAnyAddress: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1088 """Validate an IPv4 or IPv6 address. 

1089 

1090 ```python 

1091 from pydantic import BaseModel 

1092 from pydantic.networks import IPvAnyAddress 

1093 

1094 class IpModel(BaseModel): 

1095 ip: IPvAnyAddress 

1096 

1097 print(IpModel(ip='127.0.0.1')) 

1098 #> ip=IPv4Address('127.0.0.1') 

1099 

1100 try: 

1101 IpModel(ip='http://www.example.com') 

1102 except ValueError as e: 

1103 print(e.errors()) 

1104 ''' 

1105 [ 

1106 { 

1107 'type': 'ip_any_address', 

1108 'loc': ('ip',), 

1109 'msg': 'value is not a valid IPv4 or IPv6 address', 

1110 'input': 'http://www.example.com', 

1111 } 

1112 ] 

1113 ''' 

1114 ``` 

1115 """ 

1116 

1117 __slots__ = () 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1118 

1119 def __new__(cls, value: Any) -> IPvAnyAddressType: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1120 """Validate an IPv4 or IPv6 address.""" 

1121 try: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1122 return IPv4Address(value) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1123 except ValueError: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1124 pass 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1125 

1126 try: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1127 return IPv6Address(value) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1128 except ValueError: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1129 raise PydanticCustomError('ip_any_address', 'value is not a valid IPv4 or IPv6 address') 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1130 

1131 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1132 def __get_pydantic_json_schema__( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1133 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler 

1134 ) -> JsonSchemaValue: 

1135 field_schema = {} 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1136 field_schema.update(type='string', format='ipvanyaddress') 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1137 return field_schema 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1138 

1139 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1140 def __get_pydantic_core_schema__( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1141 cls, 

1142 _source: type[Any], 

1143 _handler: GetCoreSchemaHandler, 

1144 ) -> core_schema.CoreSchema: 

1145 return core_schema.no_info_plain_validator_function( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1146 cls._validate, serialization=core_schema.to_string_ser_schema() 

1147 ) 

1148 

1149 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1150 def _validate(cls, input_value: Any, /) -> IPvAnyAddressType: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1151 return cls(input_value) # type: ignore[return-value] 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1152 

1153 class IPvAnyInterface: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1154 """Validate an IPv4 or IPv6 interface.""" 

1155 

1156 __slots__ = () 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1157 

1158 def __new__(cls, value: NetworkType) -> IPvAnyInterfaceType: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1159 """Validate an IPv4 or IPv6 interface.""" 

1160 try: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1161 return IPv4Interface(value) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1162 except ValueError: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1163 pass 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1164 

1165 try: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1166 return IPv6Interface(value) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1167 except ValueError: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1168 raise PydanticCustomError('ip_any_interface', 'value is not a valid IPv4 or IPv6 interface') 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1169 

1170 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1171 def __get_pydantic_json_schema__( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1172 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler 

1173 ) -> JsonSchemaValue: 

1174 field_schema = {} 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1175 field_schema.update(type='string', format='ipvanyinterface') 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1176 return field_schema 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1177 

1178 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1179 def __get_pydantic_core_schema__( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1180 cls, 

1181 _source: type[Any], 

1182 _handler: GetCoreSchemaHandler, 

1183 ) -> core_schema.CoreSchema: 

1184 return core_schema.no_info_plain_validator_function( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1185 cls._validate, serialization=core_schema.to_string_ser_schema() 

1186 ) 

1187 

1188 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1189 def _validate(cls, input_value: NetworkType, /) -> IPvAnyInterfaceType: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1190 return cls(input_value) # type: ignore[return-value] 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1191 

1192 class IPvAnyNetwork: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1193 """Validate an IPv4 or IPv6 network.""" 

1194 

1195 __slots__ = () 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1196 

1197 def __new__(cls, value: NetworkType) -> IPvAnyNetworkType: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1198 """Validate an IPv4 or IPv6 network.""" 

1199 # Assume IP Network is defined with a default value for `strict` argument. 

1200 # Define your own class if you want to specify network address check strictness. 

1201 try: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1202 return IPv4Network(value) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1203 except ValueError: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1204 pass 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1205 

1206 try: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1207 return IPv6Network(value) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1208 except ValueError: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1209 raise PydanticCustomError('ip_any_network', 'value is not a valid IPv4 or IPv6 network') 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1210 

1211 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1212 def __get_pydantic_json_schema__( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1213 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler 

1214 ) -> JsonSchemaValue: 

1215 field_schema = {} 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1216 field_schema.update(type='string', format='ipvanynetwork') 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1217 return field_schema 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1218 

1219 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1220 def __get_pydantic_core_schema__( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1221 cls, 

1222 _source: type[Any], 

1223 _handler: GetCoreSchemaHandler, 

1224 ) -> core_schema.CoreSchema: 

1225 return core_schema.no_info_plain_validator_function( 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1226 cls._validate, serialization=core_schema.to_string_ser_schema() 

1227 ) 

1228 

1229 @classmethod 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1230 def _validate(cls, input_value: NetworkType, /) -> IPvAnyNetworkType: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1231 return cls(input_value) # type: ignore[return-value] 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1232 

1233 

1234def _build_pretty_email_regex() -> re.Pattern[str]: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1235 name_chars = r'[\w!#$%&\'*+\-/=?^_`{|}~]' 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1236 unquoted_name_group = rf'((?:{name_chars}+\s+)*{name_chars}+)' 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1237 quoted_name_group = r'"((?:[^"]|\")+)"' 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1238 email_group = r'<(.+)>' 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1239 return re.compile(rf'\s*(?:{unquoted_name_group}|{quoted_name_group})?\s*{email_group}\s*') 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1240 

1241 

1242pretty_email_regex = _build_pretty_email_regex() 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1243 

1244MAX_EMAIL_LENGTH = 2048 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1245"""Maximum length for an email. 1brcsdteuafvgwhxiyjzkAlBmC

1246A somewhat arbitrary but very generous number compared to what is allowed by most implementations. 

1247""" 

1248 

1249 

1250def validate_email(value: str) -> tuple[str, str]: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1251 """Email address validation using [email-validator](https://pypi.org/project/email-validator/). 

1252 

1253 Returns: 

1254 A tuple containing the local part of the email (or the name for "pretty" email addresses) 

1255 and the normalized email. 

1256 

1257 Raises: 

1258 PydanticCustomError: If the email is invalid. 

1259 

1260 Note: 

1261 Note that: 

1262 

1263 * Raw IP address (literal) domain parts are not allowed. 

1264 * `"John Doe <local_part@domain.com>"` style "pretty" email addresses are processed. 

1265 * Spaces are striped from the beginning and end of addresses, but no error is raised. 

1266 """ 

1267 if email_validator is None: 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1268 import_email_validator() 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC

1269 

1270 if len(value) > MAX_EMAIL_LENGTH: 1nbcdeoapfghiqjklm

1271 raise PydanticCustomError( 1nbcdeoapfghiqjklm

1272 'value_error', 

1273 'value is not a valid email address: {reason}', 

1274 {'reason': f'Length must not exceed {MAX_EMAIL_LENGTH} characters'}, 

1275 ) 

1276 

1277 m = pretty_email_regex.fullmatch(value) 1nbcdeoapfghiqjklm

1278 name: str | None = None 1nbcdeoapfghiqjklm

1279 if m: 1nbcdeoapfghiqjklm

1280 unquoted_name, quoted_name, value = m.groups() 1nbcdeoapfghiqjklm

1281 name = unquoted_name or quoted_name 1nbcdeoapfghiqjklm

1282 

1283 email = value.strip() 1nbcdeoapfghiqjklm

1284 

1285 try: 1nbcdeoapfghiqjklm

1286 parts = email_validator.validate_email(email, check_deliverability=False) 1nbcdeoapfghiqjklm

1287 except email_validator.EmailNotValidError as e: 1nbcdeoapfghiqjklm

1288 raise PydanticCustomError( 1nbcdeoapfghiqjklm

1289 'value_error', 'value is not a valid email address: {reason}', {'reason': str(e.args[0])} 

1290 ) from e 

1291 

1292 email = parts.normalized 1nbcdeoapfghiqjklm

1293 assert email is not None 1nbcdeoapfghiqjklm

1294 name = name or parts.local_part 1nbcdeoapfghiqjklm

1295 return name, email 1nbcdeoapfghiqjklm

1296 

1297 

1298__getattr__ = getattr_migration(__name__) 1nDbrcsdteuoapEfvgwhxiyqFjzkAlBmC