Coverage for pydantic/networks.py: 92.29%

383 statements  

« prev     ^ index     » next       coverage.py v7.9.2, created at 2025-07-20 16:49 +0000

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

2 

3from __future__ import annotations as _annotations 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

4 

5import dataclasses as _dataclasses 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

6import re 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

7from dataclasses import fields 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

8from functools import lru_cache 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

9from importlib.metadata import version 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

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

12 

13from pydantic_core import ( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

14 MultiHostHost, 

15 PydanticCustomError, 

16 PydanticSerializationUnexpectedValue, 

17 SchemaSerializer, 

18 core_schema, 

19) 

20from pydantic_core import MultiHostUrl as _CoreMultiHostUrl 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

21from pydantic_core import Url as _CoreUrl 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

22from typing_extensions import Self, TypeAlias 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

23 

24from pydantic.errors import PydanticUserError 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

25 

26from ._internal import _repr, _schema_generation_shared 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

27from ._migration import getattr_migration 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

28from .annotated_handlers import GetCoreSchemaHandler 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

29from .json_schema import JsonSchemaValue 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

30from .type_adapter import TypeAdapter 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

31 

32if TYPE_CHECKING: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

33 import email_validator 

34 

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

36 

37else: 

38 email_validator = None 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

39 

40 

41__all__ = [ 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

71class UrlConstraints: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

85 host_required: bool | None = None 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

86 default_host: str | None = None 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

87 default_port: int | None = None 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

88 default_path: str | None = None 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

89 

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

91 return hash( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

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} 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

106 

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

108 schema = handler(source) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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 true1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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(): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

119 schema_to_mutate[constraint_key] = constraint_value 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

120 return schema 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

121 

122 

123class _BaseUrl: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

125 _url: _CoreUrl 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

126 

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

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

129 

130 @property 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

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 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

137 

138 @property 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

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 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

145 

146 @property 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

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 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

153 

154 @property 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

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 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

162 

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

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 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

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 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

180 

181 @property 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

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 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

188 

189 @property 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

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 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

196 

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

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 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

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 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

211 

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

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() 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

219 

220 def encoded_string(self) -> str: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

221 """The URL's encoded string representation via __str__(). 

222 

223 This returns the punycode-encoded host version of the URL as a string. 

224 """ 

225 return str(self) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

226 

227 def __str__(self) -> str: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

229 return str(self._url) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

230 

231 def __repr__(self) -> str: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

233 

234 def __deepcopy__(self, memo: dict) -> Self: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

235 return self.__class__(self._url) 

236 

237 def __eq__(self, other: Any) -> bool: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

238 return self.__class__ is other.__class__ and self._url == other._url 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

239 

240 def __lt__(self, other: Any) -> bool: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

241 return self.__class__ is other.__class__ and self._url < other._url 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

242 

243 def __gt__(self, other: Any) -> bool: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

244 return self.__class__ is other.__class__ and self._url > other._url 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

245 

246 def __le__(self, other: Any) -> bool: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

247 return self.__class__ is other.__class__ and self._url <= other._url 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

248 

249 def __ge__(self, other: Any) -> bool: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

250 return self.__class__ is other.__class__ and self._url >= other._url 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

251 

252 def __hash__(self) -> int: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

253 return hash(self._url) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

254 

255 def __len__(self) -> int: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

256 return len(str(self._url)) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

257 

258 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

259 def build( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

260 cls, 

261 *, 

262 scheme: str, 

263 username: str | None = None, 

264 password: str | None = None, 

265 host: str, 

266 port: int | None = None, 

267 path: str | None = None, 

268 query: str | None = None, 

269 fragment: str | None = None, 

270 ) -> Self: 

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

272 

273 Args: 

274 scheme: The scheme part of the URL. 

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

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

277 host: The host part of the URL. 

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

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

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

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

282 

283 Returns: 

284 An instance of URL 

285 """ 

286 return cls( 

287 _CoreUrl.build( 

288 scheme=scheme, 

289 username=username, 

290 password=password, 

291 host=host, 

292 port=port, 

293 path=path, 

294 query=query, 

295 fragment=fragment, 

296 ) 

297 ) 

298 

299 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

301 if not isinstance(url, cls): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

302 raise PydanticSerializationUnexpectedValue( 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

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

304 ) 

305 if info.mode == 'json': 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

306 return str(url) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

307 return url 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

308 

309 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

310 def __get_pydantic_core_schema__( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

312 ) -> core_schema.CoreSchema: 

313 def wrap_val(v, h): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

314 if isinstance(v, source): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

315 return v 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

316 if isinstance(v, _BaseUrl): 316 ↛ 317line 316 didn't jump to line 317 because the condition on line 316 was never true1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

317 v = str(v) 

318 core_url = h(v) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

319 instance = source.__new__(source) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

320 instance._url = core_url 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

321 return instance 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

322 

323 return core_schema.no_info_wrap_validator_function( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

324 wrap_val, 

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

326 serialization=core_schema.plain_serializer_function_ser_schema( 

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

328 ), 

329 ) 

330 

331 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

332 def __get_pydantic_json_schema__( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

334 ) -> JsonSchemaValue: 

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

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

337 inner_schema = core_schema['schema'] if core_schema['type'] == 'function-wrap' else core_schema 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

338 return handler(inner_schema) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

339 

340 __pydantic_serializer__ = SchemaSerializer(core_schema.any_schema(serialization=core_schema.to_string_ser_schema())) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

341 

342 

343class _BaseMultiHostUrl: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

344 _constraints: ClassVar[UrlConstraints] = UrlConstraints() 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

345 _url: _CoreMultiHostUrl 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

346 

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

348 self._url = _build_type_adapter(self.__class__).validate_python(url)._url 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

349 

350 @property 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

351 def scheme(self) -> str: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

352 """The scheme part of the URL. 

353 

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

355 """ 

356 return self._url.scheme 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

357 

358 @property 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

359 def path(self) -> str | None: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

361 

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

363 """ 

364 return self._url.path 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

365 

366 @property 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

367 def query(self) -> str | None: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

369 

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

371 """ 

372 return self._url.query 

373 

374 def query_params(self) -> list[tuple[str, str]]: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

376 

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

378 """ 

379 return self._url.query_params() 

380 

381 @property 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

382 def fragment(self) -> str | None: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

384 

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

386 """ 

387 return self._url.fragment 

388 

389 def hosts(self) -> list[MultiHostHost]: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

391 

392 ```python 

393 from pydantic_core import MultiHostUrl 

394 

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

396 print(mhu.hosts()) 

397 """ 

398 [ 

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

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

401 ] 

402 ``` 

403 Returns: 

404 A list of dicts, each representing a host. 

405 ''' 

406 return self._url.hosts() 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

407 

408 def encoded_string(self) -> str: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

409 """The URL's encoded string representation via __str__(). 

410 

411 This returns the punycode-encoded host version of the URL as a string. 

412 """ 

413 return str(self) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

414 

415 def unicode_string(self) -> str: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

417 return self._url.unicode_string() 

418 

419 def __str__(self) -> str: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

421 return str(self._url) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

422 

423 def __repr__(self) -> str: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

425 

426 def __deepcopy__(self, memo: dict) -> Self: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

427 return self.__class__(self._url) 

428 

429 def __eq__(self, other: Any) -> bool: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

430 return self.__class__ is other.__class__ and self._url == other._url 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

431 

432 def __hash__(self) -> int: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

433 return hash(self._url) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

434 

435 def __len__(self) -> int: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

436 return len(str(self._url)) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

437 

438 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

439 def build( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

440 cls, 

441 *, 

442 scheme: str, 

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

444 username: str | None = None, 

445 password: str | None = None, 

446 host: str | None = None, 

447 port: int | None = None, 

448 path: str | None = None, 

449 query: str | None = None, 

450 fragment: str | None = None, 

451 ) -> Self: 

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

453 

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

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

456 

457 Args: 

458 scheme: The scheme part of the URL. 

459 hosts: Multiple hosts to build the URL from. 

460 username: The username part of the URL. 

461 password: The password part of the URL. 

462 host: The host part of the URL. 

463 port: The port part of the URL. 

464 path: The path part of the URL. 

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

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

467 

468 Returns: 

469 An instance of `MultiHostUrl` 

470 """ 

471 return cls( 

472 _CoreMultiHostUrl.build( 

473 scheme=scheme, 

474 hosts=hosts, 

475 username=username, 

476 password=password, 

477 host=host, 

478 port=port, 

479 path=path, 

480 query=query, 

481 fragment=fragment, 

482 ) 

483 ) 

484 

485 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

487 if not isinstance(url, cls): 

488 raise PydanticSerializationUnexpectedValue( 

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

490 ) 

491 if info.mode == 'json': 

492 return str(url) 

493 return url 

494 

495 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

496 def __get_pydantic_core_schema__( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

498 ) -> core_schema.CoreSchema: 

499 def wrap_val(v, h): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

500 if isinstance(v, source): 500 ↛ 501line 500 didn't jump to line 501 because the condition on line 500 was never true1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

501 return v 

502 if isinstance(v, _BaseMultiHostUrl): 502 ↛ 503line 502 didn't jump to line 503 because the condition on line 502 was never true1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

503 v = str(v) 

504 core_url = h(v) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

505 instance = source.__new__(source) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

506 instance._url = core_url 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

507 return instance 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

508 

509 return core_schema.no_info_wrap_validator_function( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

510 wrap_val, 

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

512 serialization=core_schema.plain_serializer_function_ser_schema( 

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

514 ), 

515 ) 

516 

517 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

518 def __get_pydantic_json_schema__( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

520 ) -> JsonSchemaValue: 

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

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

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

524 return handler(inner_schema) 

525 

526 __pydantic_serializer__ = SchemaSerializer(core_schema.any_schema(serialization=core_schema.to_string_ser_schema())) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

527 

528 

529@lru_cache 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

531 return TypeAdapter(cls) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

532 

533 

534class AnyUrl(_BaseUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

535 """Base type for all URLs. 

536 

537 * Any scheme allowed 

538 * Top-level domain (TLD) not required 

539 * Host not required 

540 

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

542 the types export the following properties: 

543 

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

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

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

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

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

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

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

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

552 """ 

553 

554 

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

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

557 

558 

559class AnyHttpUrl(AnyUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

561 

562 * TLD not required 

563 * Host not required 

564 """ 

565 

566 _constraints = UrlConstraints(allowed_schemes=['http', 'https']) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

567 

568 

569class HttpUrl(AnyUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

571 

572 * TLD not required 

573 * Host not required 

574 * Max length 2083 

575 

576 ```python 

577 from pydantic import BaseModel, HttpUrl, ValidationError 

578 

579 class MyModel(BaseModel): 

580 url: HttpUrl 

581 

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

583 print(m.url) 

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

585 

586 try: 

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

588 except ValidationError as e: 

589 print(e) 

590 ''' 

591 1 validation error for MyModel 

592 url 

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

594 ''' 

595 

596 try: 

597 MyModel(url='not a url') 

598 except ValidationError as e: 

599 print(e) 

600 ''' 

601 1 validation error for MyModel 

602 url 

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

604 ''' 

605 ``` 

606 

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

608 

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

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

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

612 

613 ```python 

614 from pydantic import BaseModel, HttpUrl 

615 

616 class MyModel(BaseModel): 

617 url: HttpUrl 

618 

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

620 print(m1.url) 

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

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

623 print(m2.url) 

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

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

626 print(m3.url) 

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

628 ``` 

629 

630 

631 !!! warning "Underscores in Hostnames" 

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

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

634 

635 To explain this; consider the following two cases: 

636 

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

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

639 

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

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

642 

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

644 (or at least big) company. 

645 """ 

646 

647 _constraints = UrlConstraints(max_length=2083, allowed_schemes=['http', 'https']) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

648 

649 

650class AnyWebsocketUrl(AnyUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

652 

653 * TLD not required 

654 * Host not required 

655 """ 

656 

657 _constraints = UrlConstraints(allowed_schemes=['ws', 'wss']) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

658 

659 

660class WebsocketUrl(AnyUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

662 

663 * TLD not required 

664 * Host not required 

665 * Max length 2083 

666 """ 

667 

668 _constraints = UrlConstraints(max_length=2083, allowed_schemes=['ws', 'wss']) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

669 

670 

671class FileUrl(AnyUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

673 

674 * Host not required 

675 """ 

676 

677 _constraints = UrlConstraints(allowed_schemes=['file']) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

678 

679 

680class FtpUrl(AnyUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

682 

683 * TLD not required 

684 * Host not required 

685 """ 

686 

687 _constraints = UrlConstraints(allowed_schemes=['ftp']) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

688 

689 

690class PostgresDsn(_BaseMultiHostUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

692 

693 * User info required 

694 * TLD not required 

695 * Host required 

696 * Supports multiple hosts 

697 

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

699 

700 ```python 

701 from pydantic import ( 

702 BaseModel, 

703 HttpUrl, 

704 PostgresDsn, 

705 ValidationError, 

706 field_validator, 

707 ) 

708 

709 class MyModel(BaseModel): 

710 url: HttpUrl 

711 

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

713 

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

715 print(repr(m.url)) 

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

717 print(m.url.scheme) 

718 #> http 

719 print(m.url.host) 

720 #> www.example.com 

721 print(m.url.port) 

722 #> 80 

723 

724 class MyDatabaseModel(BaseModel): 

725 db: PostgresDsn 

726 

727 @field_validator('db') 

728 def check_db_name(cls, v): 

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

730 return v 

731 

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

733 print(m.db) 

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

735 

736 try: 

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

738 except ValidationError as e: 

739 print(e) 

740 ''' 

741 1 validation error for MyDatabaseModel 

742 db 

743 Assertion failed, database must be provided 

744 assert (None) 

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

746 ''' 

747 ``` 

748 """ 

749 

750 _constraints = UrlConstraints( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

751 host_required=True, 

752 allowed_schemes=[ 

753 'postgres', 

754 'postgresql', 

755 'postgresql+asyncpg', 

756 'postgresql+pg8000', 

757 'postgresql+psycopg', 

758 'postgresql+psycopg2', 

759 'postgresql+psycopg2cffi', 

760 'postgresql+py-postgresql', 

761 'postgresql+pygresql', 

762 ], 

763 ) 

764 

765 @property 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

766 def host(self) -> str: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

767 """The required URL host.""" 

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

769 

770 

771class CockroachDsn(AnyUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

773 

774 * User info required 

775 * TLD not required 

776 * Host required 

777 """ 

778 

779 _constraints = UrlConstraints( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

780 host_required=True, 

781 allowed_schemes=[ 

782 'cockroachdb', 

783 'cockroachdb+psycopg2', 

784 'cockroachdb+asyncpg', 

785 ], 

786 ) 

787 

788 @property 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

789 def host(self) -> str: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

790 """The required URL host.""" 

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

792 

793 

794class AmqpDsn(AnyUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

796 

797 * User info required 

798 * TLD not required 

799 * Host not required 

800 """ 

801 

802 _constraints = UrlConstraints(allowed_schemes=['amqp', 'amqps']) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

803 

804 

805class RedisDsn(AnyUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

807 

808 * User info required 

809 * TLD not required 

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

811 """ 

812 

813 _constraints = UrlConstraints( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

815 default_host='localhost', 

816 default_port=6379, 

817 default_path='/0', 

818 host_required=True, 

819 ) 

820 

821 @property 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

822 def host(self) -> str: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

823 """The required URL host.""" 

824 return self._url.host # pyright: ignore[reportReturnType] 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

825 

826 

827class MongoDsn(_BaseMultiHostUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

829 

830 * User info not required 

831 * Database name not required 

832 * Port not required 

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

834 """ 

835 

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

837 

838 

839class KafkaDsn(AnyUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

841 

842 * User info required 

843 * TLD not required 

844 * Host not required 

845 """ 

846 

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

848 

849 

850class NatsDsn(_BaseMultiHostUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

852 

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

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

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

856 More: https://nats.io 

857 """ 

858 

859 _constraints = UrlConstraints( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

861 ) 

862 

863 

864class MySQLDsn(AnyUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

866 

867 * User info required 

868 * TLD not required 

869 * Host not required 

870 """ 

871 

872 _constraints = UrlConstraints( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

873 allowed_schemes=[ 

874 'mysql', 

875 'mysql+mysqlconnector', 

876 'mysql+aiomysql', 

877 'mysql+asyncmy', 

878 'mysql+mysqldb', 

879 'mysql+pymysql', 

880 'mysql+cymysql', 

881 'mysql+pyodbc', 

882 ], 

883 default_port=3306, 

884 host_required=True, 

885 ) 

886 

887 

888class MariaDBDsn(AnyUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

890 

891 * User info required 

892 * TLD not required 

893 * Host not required 

894 """ 

895 

896 _constraints = UrlConstraints( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

898 default_port=3306, 

899 ) 

900 

901 

902class ClickHouseDsn(AnyUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

904 

905 * User info required 

906 * TLD not required 

907 * Host not required 

908 """ 

909 

910 _constraints = UrlConstraints( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

911 allowed_schemes=[ 

912 'clickhouse+native', 

913 'clickhouse+asynch', 

914 'clickhouse+http', 

915 'clickhouse', 

916 'clickhouses', 

917 'clickhousedb', 

918 ], 

919 default_host='localhost', 

920 default_port=9000, 

921 ) 

922 

923 

924class SnowflakeDsn(AnyUrl): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

926 

927 * User info required 

928 * TLD not required 

929 * Host required 

930 """ 

931 

932 _constraints = UrlConstraints( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

933 allowed_schemes=['snowflake'], 

934 host_required=True, 

935 ) 

936 

937 @property 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

938 def host(self) -> str: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

939 """The required URL host.""" 

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

941 

942 

943def import_email_validator() -> None: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

944 global email_validator 

945 try: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

946 import email_validator 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

947 except ImportError as e: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

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

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

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

951 

952 

953if TYPE_CHECKING: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

954 EmailStr = Annotated[str, ...] 

955else: 

956 

957 class EmailStr: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

958 """ 

959 Info: 

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

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

962 

963 ```bash 

964 pip install email-validator 

965 ``` 

966 

967 Validate email addresses. 

968 

969 ```python 

970 from pydantic import BaseModel, EmailStr 

971 

972 class Model(BaseModel): 

973 email: EmailStr 

974 

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

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

977 ``` 

978 """ # noqa: D212 

979 

980 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

981 def __get_pydantic_core_schema__( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

982 cls, 

983 _source: type[Any], 

984 _handler: GetCoreSchemaHandler, 

985 ) -> core_schema.CoreSchema: 

986 import_email_validator() 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

987 return core_schema.no_info_after_validator_function(cls._validate, core_schema.str_schema()) 1qbcdefrasghijktlmnop

988 

989 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

990 def __get_pydantic_json_schema__( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

992 ) -> JsonSchemaValue: 

993 field_schema = handler(core_schema) 1qbcdefrasghijktlmnop

994 field_schema.update(type='string', format='email') 1qbcdefrasghijktlmnop

995 return field_schema 1qbcdefrasghijktlmnop

996 

997 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

998 def _validate(cls, input_value: str, /) -> str: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

999 return validate_email(input_value)[1] 1qbcdefrasghijktlmnop

1000 

1001 

1002class NameEmail(_repr.Representation): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1003 """ 

1004 Info: 

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

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

1007 

1008 ```bash 

1009 pip install email-validator 

1010 ``` 

1011 

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

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

1014 

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

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

1017 

1018 ```python 

1019 from pydantic import BaseModel, NameEmail 

1020 

1021 class User(BaseModel): 

1022 email: NameEmail 

1023 

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

1025 print(user.email) 

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

1027 print(user.email.name) 

1028 #> Fred Bloggs 

1029 

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

1031 print(user.email) 

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

1033 print(user.email.name) 

1034 #> fred.bloggs 

1035 ``` 

1036 """ # noqa: D212 

1037 

1038 __slots__ = 'name', 'email' 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1039 

1040 def __init__(self, name: str, email: str): 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1041 self.name = name 1qbcdefrasghijktlmnop

1042 self.email = email 1qbcdefrasghijktlmnop

1043 

1044 def __eq__(self, other: Any) -> bool: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1045 return isinstance(other, NameEmail) and (self.name, self.email) == (other.name, other.email) 1qbcdefrasghijktlmnop

1046 

1047 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1048 def __get_pydantic_json_schema__( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

1050 ) -> JsonSchemaValue: 

1051 field_schema = handler(core_schema) 1qbcdefrasghijktlmnop

1052 field_schema.update(type='string', format='name-email') 1qbcdefrasghijktlmnop

1053 return field_schema 1qbcdefrasghijktlmnop

1054 

1055 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1056 def __get_pydantic_core_schema__( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1057 cls, 

1058 _source: type[Any], 

1059 _handler: GetCoreSchemaHandler, 

1060 ) -> core_schema.CoreSchema: 

1061 import_email_validator() 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1062 

1063 return core_schema.no_info_after_validator_function( 1qbcdefrasghijktlmnop

1064 cls._validate, 

1065 core_schema.json_or_python_schema( 

1066 json_schema=core_schema.str_schema(), 

1067 python_schema=core_schema.union_schema( 

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

1069 custom_error_type='name_email_type', 

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

1071 ), 

1072 serialization=core_schema.to_string_ser_schema(), 

1073 ), 

1074 ) 

1075 

1076 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1077 def _validate(cls, input_value: Self | str, /) -> Self: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1078 if isinstance(input_value, str): 1qbcdefrasghijktlmnop

1079 name, email = validate_email(input_value) 1qbcdefrasghijktlmnop

1080 return cls(name, email) 1qbcdefrasghijktlmnop

1081 else: 

1082 return input_value 1qbcdefrasghijktlmnop

1083 

1084 def __str__(self) -> str: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1085 if '@' in self.name: 1qbcdefrasghijktlmnop

1086 return f'"{self.name}" <{self.email}>' 1qbcdefrasghijktlmnop

1087 

1088 return f'{self.name} <{self.email}>' 1qbcdefrasghijktlmnop

1089 

1090 

1091IPvAnyAddressType: TypeAlias = 'IPv4Address | IPv6Address' 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1092IPvAnyInterfaceType: TypeAlias = 'IPv4Interface | IPv6Interface' 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1093IPvAnyNetworkType: TypeAlias = 'IPv4Network | IPv6Network' 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1094 

1095if TYPE_CHECKING: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1096 IPvAnyAddress = IPvAnyAddressType 

1097 IPvAnyInterface = IPvAnyInterfaceType 

1098 IPvAnyNetwork = IPvAnyNetworkType 

1099else: 

1100 

1101 class IPvAnyAddress: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1102 """Validate an IPv4 or IPv6 address. 

1103 

1104 ```python 

1105 from pydantic import BaseModel 

1106 from pydantic.networks import IPvAnyAddress 

1107 

1108 class IpModel(BaseModel): 

1109 ip: IPvAnyAddress 

1110 

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

1112 #> ip=IPv4Address('127.0.0.1') 

1113 

1114 try: 

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

1116 except ValueError as e: 

1117 print(e.errors()) 

1118 ''' 

1119 [ 

1120 { 

1121 'type': 'ip_any_address', 

1122 'loc': ('ip',), 

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

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

1125 } 

1126 ] 

1127 ''' 

1128 ``` 

1129 """ 

1130 

1131 __slots__ = () 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1132 

1133 def __new__(cls, value: Any) -> IPvAnyAddressType: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

1135 try: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1136 return IPv4Address(value) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1137 except ValueError: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

1138 pass 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

1139 

1140 try: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

1141 return IPv6Address(value) 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

1142 except ValueError: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

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

1144 

1145 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1146 def __get_pydantic_json_schema__( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

1148 ) -> JsonSchemaValue: 

1149 field_schema = {} 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1150 field_schema.update(type='string', format='ipvanyaddress') 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1151 return field_schema 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1152 

1153 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1154 def __get_pydantic_core_schema__( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1155 cls, 

1156 _source: type[Any], 

1157 _handler: GetCoreSchemaHandler, 

1158 ) -> core_schema.CoreSchema: 

1159 return core_schema.no_info_plain_validator_function( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

1161 ) 

1162 

1163 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1164 def _validate(cls, input_value: Any, /) -> IPvAnyAddressType: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1165 return cls(input_value) # type: ignore[return-value] 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1166 

1167 class IPvAnyInterface: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

1169 

1170 __slots__ = () 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1171 

1172 def __new__(cls, value: NetworkType) -> IPvAnyInterfaceType: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

1174 try: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1175 return IPv4Interface(value) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1176 except ValueError: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

1177 pass 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

1178 

1179 try: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

1180 return IPv6Interface(value) 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

1181 except ValueError: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

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

1183 

1184 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1185 def __get_pydantic_json_schema__( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

1187 ) -> JsonSchemaValue: 

1188 field_schema = {} 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1189 field_schema.update(type='string', format='ipvanyinterface') 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1190 return field_schema 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1191 

1192 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1193 def __get_pydantic_core_schema__( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1194 cls, 

1195 _source: type[Any], 

1196 _handler: GetCoreSchemaHandler, 

1197 ) -> core_schema.CoreSchema: 

1198 return core_schema.no_info_plain_validator_function( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

1200 ) 

1201 

1202 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1203 def _validate(cls, input_value: NetworkType, /) -> IPvAnyInterfaceType: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1204 return cls(input_value) # type: ignore[return-value] 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1205 

1206 class IPvAnyNetwork: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

1208 

1209 __slots__ = () 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1210 

1211 def __new__(cls, value: NetworkType) -> IPvAnyNetworkType: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

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

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

1215 try: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1216 return IPv4Network(value) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1217 except ValueError: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

1218 pass 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

1219 

1220 try: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

1221 return IPv6Network(value) 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

1222 except ValueError: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI

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

1224 

1225 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1226 def __get_pydantic_json_schema__( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

1228 ) -> JsonSchemaValue: 

1229 field_schema = {} 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1230 field_schema.update(type='string', format='ipvanynetwork') 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1231 return field_schema 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1232 

1233 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1234 def __get_pydantic_core_schema__( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1235 cls, 

1236 _source: type[Any], 

1237 _handler: GetCoreSchemaHandler, 

1238 ) -> core_schema.CoreSchema: 

1239 return core_schema.no_info_plain_validator_function( 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

1241 ) 

1242 

1243 @classmethod 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1244 def _validate(cls, input_value: NetworkType, /) -> IPvAnyNetworkType: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1245 return cls(input_value) # type: ignore[return-value] 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1246 

1247 

1248def _build_pretty_email_regex() -> re.Pattern[str]: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

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

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

1252 email_group = r'<(.+)>' 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

1254 

1255 

1256pretty_email_regex = _build_pretty_email_regex() 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1257 

1258MAX_EMAIL_LENGTH = 2048 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1259"""Maximum length for an email. 1bucvdwexfyMagzhAiBjCkDNlEmFnGoHpIO

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

1261""" 

1262 

1263 

1264def validate_email(value: str) -> tuple[str, str]: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

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

1266 

1267 Returns: 

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

1269 and the normalized email. 

1270 

1271 Raises: 

1272 PydanticCustomError: If the email is invalid. 

1273 

1274 Note: 

1275 Note that: 

1276 

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

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

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

1280 """ 

1281 if email_validator is None: 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1282 import_email_validator() 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO

1283 

1284 if len(value) > MAX_EMAIL_LENGTH: 1qbcdefrasghijktlmnop

1285 raise PydanticCustomError( 1qbcdefrasghijktlmnop

1286 'value_error', 

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

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

1289 ) 

1290 

1291 m = pretty_email_regex.fullmatch(value) 1qbcdefrasghijktlmnop

1292 name: str | None = None 1qbcdefrasghijktlmnop

1293 if m: 1qbcdefrasghijktlmnop

1294 unquoted_name, quoted_name, value = m.groups() 1qbcdefrasghijktlmnop

1295 name = unquoted_name or quoted_name 1qbcdefrasghijktlmnop

1296 

1297 email = value.strip() 1qbcdefrasghijktlmnop

1298 

1299 try: 1qbcdefrasghijktlmnop

1300 parts = email_validator.validate_email(email, check_deliverability=False) 1qbcdefrasghijktlmnop

1301 except email_validator.EmailNotValidError as e: 1qbcdefrasghijktlmnop

1302 raise PydanticCustomError( 1qbcdefrasghijktlmnop

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

1304 ) from e 

1305 

1306 email = parts.normalized 1qbcdefrasghijktlmnop

1307 assert email is not None 1qbcdefrasghijktlmnop

1308 name = name or parts.local_part 1qbcdefrasghijktlmnop

1309 return name, email 1qbcdefrasghijktlmnop

1310 

1311 

1312__getattr__ = getattr_migration(__name__) 1qJbucvdwexfyMrasKgzhAiBjCkDNtLlEmFnGoHpIO