Coverage for pydantic/networks.py: 92.29%

383 statements  

« prev     ^ index     » next       coverage.py v7.10.0, created at 2025-07-26 11:49 +0000

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

2 

3from __future__ import annotations as _annotations 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

4 

5import dataclasses as _dataclasses 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

6import re 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

7from dataclasses import fields 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

8from functools import lru_cache 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

9from importlib.metadata import version 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

12 

13from pydantic_core import ( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

14 MultiHostHost, 

15 PydanticCustomError, 

16 PydanticSerializationUnexpectedValue, 

17 SchemaSerializer, 

18 core_schema, 

19) 

20from pydantic_core import MultiHostUrl as _CoreMultiHostUrl 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

21from pydantic_core import Url as _CoreUrl 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

22from typing_extensions import Self, TypeAlias 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

23 

24from pydantic.errors import PydanticUserError 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

25 

26from ._internal import _repr, _schema_generation_shared 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

27from ._migration import getattr_migration 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

28from .annotated_handlers import GetCoreSchemaHandler 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

29from .json_schema import JsonSchemaValue 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

30from .type_adapter import TypeAdapter 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

31 

32if TYPE_CHECKING: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

33 import email_validator 

34 

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

36 

37else: 

38 email_validator = None 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

39 

40 

41__all__ = [ 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

71class UrlConstraints: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

85 host_required: bool | None = None 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

86 default_host: str | None = None 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

87 default_port: int | None = None 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

88 default_path: str | None = None 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

89 

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

91 return hash( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

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

106 

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

108 schema = handler(source) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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 true1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

119 schema_to_mutate[constraint_key] = constraint_value 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

120 return schema 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

121 

122 

123class _BaseUrl: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

125 _url: _CoreUrl 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

126 

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

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

129 

130 @property 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

137 

138 @property 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

145 

146 @property 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

153 

154 @property 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

162 

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

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

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

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

180 

181 @property 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

188 

189 @property 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

196 

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

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

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

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

211 

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

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

219 

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

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) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

226 

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

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

229 return str(self._url) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

230 

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

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

233 

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

235 return self.__class__(self._url) 

236 

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

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

239 

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

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

242 

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

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

245 

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

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

248 

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

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

251 

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

253 return hash(self._url) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

254 

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

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

257 

258 @classmethod 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

259 def build( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

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

302 raise PydanticSerializationUnexpectedValue( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

304 ) 

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

306 return str(url) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

307 return url 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

308 

309 @classmethod 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

310 def __get_pydantic_core_schema__( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

312 ) -> core_schema.CoreSchema: 

313 def wrap_val(v, h): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

314 if isinstance(v, source): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

315 return v 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

317 v = str(v) 

318 core_url = h(v) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

319 instance = source.__new__(source) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

320 instance._url = core_url 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

321 return instance 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

322 

323 return core_schema.no_info_wrap_validator_function( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

332 def __get_pydantic_json_schema__( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

338 return handler(inner_schema) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

339 

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

341 

342 

343class _BaseMultiHostUrl: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

345 _url: _CoreMultiHostUrl 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

346 

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

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

349 

350 @property 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

357 

358 @property 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

365 

366 @property 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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]]: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

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]: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

407 

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

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) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

414 

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

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: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

421 return str(self._url) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

422 

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

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

425 

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

427 return self.__class__(self._url) 

428 

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

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

431 

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

433 return hash(self._url) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

434 

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

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

437 

438 @classmethod 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

439 def build( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

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

496 def __get_pydantic_core_schema__( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

498 ) -> core_schema.CoreSchema: 

499 def wrap_val(v, h): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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 true1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

503 v = str(v) 

504 core_url = h(v) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

505 instance = source.__new__(source) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

506 instance._url = core_url 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

507 return instance 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

508 

509 return core_schema.no_info_wrap_validator_function( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

518 def __get_pydantic_json_schema__( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

527 

528 

529@lru_cache 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

531 return TypeAdapter(cls) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

532 

533 

534class AnyUrl(_BaseUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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']) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

567 

568 

569class HttpUrl(AnyUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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']) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

648 

649 

650class AnyWebsocketUrl(AnyUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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']) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

658 

659 

660class WebsocketUrl(AnyUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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']) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

669 

670 

671class FileUrl(AnyUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

673 

674 * Host not required 

675 """ 

676 

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

678 

679 

680class FtpUrl(AnyUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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']) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

688 

689 

690class PostgresDsn(_BaseMultiHostUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

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

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

769 

770 

771class CockroachDsn(AnyUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

780 host_required=True, 

781 allowed_schemes=[ 

782 'cockroachdb', 

783 'cockroachdb+psycopg2', 

784 'cockroachdb+asyncpg', 

785 ], 

786 ) 

787 

788 @property 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

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

792 

793 

794class AmqpDsn(AnyUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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']) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

803 

804 

805class RedisDsn(AnyUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

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

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

825 

826 

827class MongoDsn(_BaseMultiHostUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

837 

838 

839class KafkaDsn(AnyUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

848 

849 

850class NatsDsn(_BaseMultiHostUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

861 ) 

862 

863 

864class MySQLDsn(AnyUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

898 default_port=3306, 

899 ) 

900 

901 

902class ClickHouseDsn(AnyUrl): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

933 allowed_schemes=['snowflake'], 

934 host_required=True, 

935 ) 

936 

937 @property 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

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

941 

942 

943def import_email_validator() -> None: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

944 global email_validator 

945 try: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

946 import email_validator 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

947 except ImportError as e: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

954 EmailStr = Annotated[str, ...] 

955else: 

956 

957 class EmailStr: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

981 def __get_pydantic_core_schema__( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

982 cls, 

983 _source: type[Any], 

984 _handler: GetCoreSchemaHandler, 

985 ) -> core_schema.CoreSchema: 

986 import_email_validator() 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

988 

989 @classmethod 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

990 def __get_pydantic_json_schema__( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

999 return validate_email(input_value)[1] 1qbcdefrasghijktlmnop

1000 

1001 

1002class NameEmail(_repr.Representation): 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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' 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1039 

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

1041 self.name = name 1qbcdefrasghijktlmnop

1042 self.email = email 1qbcdefrasghijktlmnop

1043 

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

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

1046 

1047 @classmethod 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1048 def __get_pydantic_json_schema__( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1056 def __get_pydantic_core_schema__( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1057 cls, 

1058 _source: type[Any], 

1059 _handler: GetCoreSchemaHandler, 

1060 ) -> core_schema.CoreSchema: 

1061 import_email_validator() 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

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: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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' 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

1094 

1095if TYPE_CHECKING: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1096 IPvAnyAddress = IPvAnyAddressType 

1097 IPvAnyInterface = IPvAnyInterfaceType 

1098 IPvAnyNetwork = IPvAnyNetworkType 

1099else: 

1100 

1101 class IPvAnyAddress: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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__ = () 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1132 

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

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

1135 try: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1136 return IPv4Address(value) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1137 except ValueError: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1138 pass 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1139 

1140 try: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1141 return IPv6Address(value) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1142 except ValueError: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1144 

1145 @classmethod 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1146 def __get_pydantic_json_schema__( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1148 ) -> JsonSchemaValue: 

1149 field_schema = {} 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1151 return field_schema 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1152 

1153 @classmethod 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1154 def __get_pydantic_core_schema__( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1155 cls, 

1156 _source: type[Any], 

1157 _handler: GetCoreSchemaHandler, 

1158 ) -> core_schema.CoreSchema: 

1159 return core_schema.no_info_plain_validator_function( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1161 ) 

1162 

1163 @classmethod 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

1166 

1167 class IPvAnyInterface: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1169 

1170 __slots__ = () 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1171 

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

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

1174 try: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1175 return IPv4Interface(value) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1176 except ValueError: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1177 pass 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1178 

1179 try: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1180 return IPv6Interface(value) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1181 except ValueError: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1183 

1184 @classmethod 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1185 def __get_pydantic_json_schema__( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1187 ) -> JsonSchemaValue: 

1188 field_schema = {} 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1190 return field_schema 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1191 

1192 @classmethod 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1193 def __get_pydantic_core_schema__( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1194 cls, 

1195 _source: type[Any], 

1196 _handler: GetCoreSchemaHandler, 

1197 ) -> core_schema.CoreSchema: 

1198 return core_schema.no_info_plain_validator_function( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1200 ) 

1201 

1202 @classmethod 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

1205 

1206 class IPvAnyNetwork: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1208 

1209 __slots__ = () 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1210 

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

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: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1216 return IPv4Network(value) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1217 except ValueError: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1218 pass 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1219 

1220 try: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1221 return IPv6Network(value) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1222 except ValueError: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1224 

1225 @classmethod 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1226 def __get_pydantic_json_schema__( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1228 ) -> JsonSchemaValue: 

1229 field_schema = {} 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1231 return field_schema 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1232 

1233 @classmethod 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1234 def __get_pydantic_core_schema__( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1235 cls, 

1236 _source: type[Any], 

1237 _handler: GetCoreSchemaHandler, 

1238 ) -> core_schema.CoreSchema: 

1239 return core_schema.no_info_plain_validator_function( 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

1241 ) 

1242 

1243 @classmethod 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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

1246 

1247 

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

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

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

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

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

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

1254 

1255 

1256pretty_email_regex = _build_pretty_email_regex() 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1257 

1258MAX_EMAIL_LENGTH = 2048 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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

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]: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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: 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

1282 import_email_validator() 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL

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__) 1qMbucvdwexfyzrasNgAhBiCjDkEFtOlGmHnIoJpKL