Coverage for pydantic/networks.py: 92.29%
383 statements
« prev ^ index » next coverage.py v7.8.2, created at 2025-06-11 13:08 +0000
1"""The networks module contains types for common network-related fields."""
3from __future__ import annotations as _annotations 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
5import dataclasses as _dataclasses 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
6import re 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
7from dataclasses import fields 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
8from functools import lru_cache 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
9from importlib.metadata import version 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
10from ipaddress import IPv4Address, IPv4Interface, IPv4Network, IPv6Address, IPv6Interface, IPv6Network 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
11from typing import TYPE_CHECKING, Annotated, Any, ClassVar 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
13from pydantic_core import ( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
14 MultiHostHost,
15 PydanticCustomError,
16 PydanticSerializationUnexpectedValue,
17 SchemaSerializer,
18 core_schema,
19)
20from pydantic_core import MultiHostUrl as _CoreMultiHostUrl 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
21from pydantic_core import Url as _CoreUrl 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
22from typing_extensions import Self, TypeAlias 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
24from pydantic.errors import PydanticUserError 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
26from ._internal import _repr, _schema_generation_shared 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
27from ._migration import getattr_migration 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
28from .annotated_handlers import GetCoreSchemaHandler 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
29from .json_schema import JsonSchemaValue 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
30from .type_adapter import TypeAdapter 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
32if TYPE_CHECKING: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
33 import email_validator
35 NetworkType: TypeAlias = 'str | bytes | int | tuple[str | bytes | int, str | int]' 1a
37else:
38 email_validator = None 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
41__all__ = [ 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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]
70@_dataclasses.dataclass 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
71class UrlConstraints: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
72 """Url constraints.
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 """
83 max_length: int | None = None 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
84 allowed_schemes: list[str] | None = None 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
85 host_required: bool | None = None 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
86 default_host: str | None = None 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
87 default_port: int | None = None 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
88 default_path: str | None = None 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
90 def __hash__(self) -> int: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
91 return hash( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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 )
102 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
103 def defined_constraints(self) -> dict[str, Any]: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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} 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
107 def __get_pydantic_core_schema__(self, source: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
108 schema = handler(source) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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 true1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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(): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
119 schema_to_mutate[constraint_key] = constraint_value 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
120 return schema 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
123class _BaseUrl: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
124 _constraints: ClassVar[UrlConstraints] = UrlConstraints() 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
125 _url: _CoreUrl 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
127 def __init__(self, url: str | _CoreUrl | _BaseUrl) -> None: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
128 self._url = _build_type_adapter(self.__class__).validate_python(url)._url 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
130 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
131 def scheme(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
132 """The scheme part of the URL.
134 e.g. `https` in `https://user:pass@host:port/path?query#fragment`
135 """
136 return self._url.scheme 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
138 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
139 def username(self) -> str | None: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
140 """The username part of the URL, or `None`.
142 e.g. `user` in `https://user:pass@host:port/path?query#fragment`
143 """
144 return self._url.username 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
146 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
147 def password(self) -> str | None: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
148 """The password part of the URL, or `None`.
150 e.g. `pass` in `https://user:pass@host:port/path?query#fragment`
151 """
152 return self._url.password 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
154 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
155 def host(self) -> str | None: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
156 """The host part of the URL, or `None`.
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 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
163 def unicode_host(self) -> str | None: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
164 """The host part of the URL as a unicode string, or `None`.
166 e.g. `host` in `https://user:pass@host:port/path?query#fragment`
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()
173 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
174 def port(self) -> int | None: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
175 """The port part of the URL, or `None`.
177 e.g. `port` in `https://user:pass@host:port/path?query#fragment`
178 """
179 return self._url.port 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
181 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
182 def path(self) -> str | None: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
183 """The path part of the URL, or `None`.
185 e.g. `/path` in `https://user:pass@host:port/path?query#fragment`
186 """
187 return self._url.path 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
189 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
190 def query(self) -> str | None: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
191 """The query part of the URL, or `None`.
193 e.g. `query` in `https://user:pass@host:port/path?query#fragment`
194 """
195 return self._url.query 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
197 def query_params(self) -> list[tuple[str, str]]: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
198 """The query part of the URL as a list of key-value pairs.
200 e.g. `[('foo', 'bar')]` in `https://user:pass@host:port/path?foo=bar#fragment`
201 """
202 return self._url.query_params()
204 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
205 def fragment(self) -> str | None: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
206 """The fragment part of the URL, or `None`.
208 e.g. `fragment` in `https://user:pass@host:port/path?query#fragment`
209 """
210 return self._url.fragment 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
212 def unicode_string(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
213 """The URL as a unicode string, unlike `__str__()` this will not punycode encode the host.
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() 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
220 def encoded_string(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
221 """The URL's encoded string representation via __str__().
223 This returns the punycode-encoded host version of the URL as a string.
224 """
225 return str(self) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
227 def __str__(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
228 """The URL as a string, this will punycode encode the host if required."""
229 return str(self._url) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
231 def __repr__(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
232 return f'{self.__class__.__name__}({str(self._url)!r})' 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
234 def __deepcopy__(self, memo: dict) -> Self: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
235 return self.__class__(self._url)
237 def __eq__(self, other: Any) -> bool: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
238 return self.__class__ is other.__class__ and self._url == other._url 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
240 def __lt__(self, other: Any) -> bool: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
241 return self.__class__ is other.__class__ and self._url < other._url 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
243 def __gt__(self, other: Any) -> bool: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
244 return self.__class__ is other.__class__ and self._url > other._url 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
246 def __le__(self, other: Any) -> bool: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
247 return self.__class__ is other.__class__ and self._url <= other._url 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
249 def __ge__(self, other: Any) -> bool: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
250 return self.__class__ is other.__class__ and self._url >= other._url 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
252 def __hash__(self) -> int: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
253 return hash(self._url) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
255 def __len__(self) -> int: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
256 return len(str(self._url)) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
258 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
259 def build( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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.
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.
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 )
299 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
300 def serialize_url(cls, url: Any, info: core_schema.SerializationInfo) -> str | Self: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
301 if not isinstance(url, cls): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
302 raise PydanticSerializationUnexpectedValue( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
303 f"Expected `{cls}` but got `{type(url)}` with value `'{url}'` - serialized value may not be as expected."
304 )
305 if info.mode == 'json': 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
306 return str(url) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
307 return url 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
309 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
310 def __get_pydantic_core_schema__( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
311 cls, source: type[_BaseUrl], handler: GetCoreSchemaHandler
312 ) -> core_schema.CoreSchema:
313 def wrap_val(v, h): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
314 if isinstance(v, source): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
315 return v 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
316 if isinstance(v, _BaseUrl): 316 ↛ 317line 316 didn't jump to line 317 because the condition on line 316 was never true1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
317 v = str(v)
318 core_url = h(v) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
319 instance = source.__new__(source) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
320 instance._url = core_url 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
321 return instance 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
323 return core_schema.no_info_wrap_validator_function( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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 )
331 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
332 def __get_pydantic_json_schema__( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
338 return handler(inner_schema) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
340 __pydantic_serializer__ = SchemaSerializer(core_schema.any_schema(serialization=core_schema.to_string_ser_schema())) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
343class _BaseMultiHostUrl: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
344 _constraints: ClassVar[UrlConstraints] = UrlConstraints() 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
345 _url: _CoreMultiHostUrl 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
347 def __init__(self, url: str | _CoreMultiHostUrl | _BaseMultiHostUrl) -> None: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
348 self._url = _build_type_adapter(self.__class__).validate_python(url)._url 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
350 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
351 def scheme(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
352 """The scheme part of the URL.
354 e.g. `https` in `https://foo.com,bar.com/path?query#fragment`
355 """
356 return self._url.scheme 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
358 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
359 def path(self) -> str | None: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
360 """The path part of the URL, or `None`.
362 e.g. `/path` in `https://foo.com,bar.com/path?query#fragment`
363 """
364 return self._url.path 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
366 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
367 def query(self) -> str | None: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
368 """The query part of the URL, or `None`.
370 e.g. `query` in `https://foo.com,bar.com/path?query#fragment`
371 """
372 return self._url.query
374 def query_params(self) -> list[tuple[str, str]]: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
375 """The query part of the URL as a list of key-value pairs.
377 e.g. `[('foo', 'bar')]` in `https://foo.com,bar.com/path?foo=bar#fragment`
378 """
379 return self._url.query_params()
381 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
382 def fragment(self) -> str | None: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
383 """The fragment part of the URL, or `None`.
385 e.g. `fragment` in `https://foo.com,bar.com/path?query#fragment`
386 """
387 return self._url.fragment
389 def hosts(self) -> list[MultiHostHost]: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
390 '''The hosts of the `MultiHostUrl` as [`MultiHostHost`][pydantic_core.MultiHostHost] typed dicts.
392 ```python
393 from pydantic_core import MultiHostUrl
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() 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
408 def encoded_string(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
409 """The URL's encoded string representation via __str__().
411 This returns the punycode-encoded host version of the URL as a string.
412 """
413 return str(self) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
415 def unicode_string(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
416 """The URL as a unicode string, unlike `__str__()` this will not punycode encode the hosts."""
417 return self._url.unicode_string()
419 def __str__(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
420 """The URL as a string, this will punycode encode the host if required."""
421 return str(self._url) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
423 def __repr__(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
424 return f'{self.__class__.__name__}({str(self._url)!r})' 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
426 def __deepcopy__(self, memo: dict) -> Self: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
427 return self.__class__(self._url)
429 def __eq__(self, other: Any) -> bool: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
430 return self.__class__ is other.__class__ and self._url == other._url 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
432 def __hash__(self) -> int: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
433 return hash(self._url) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
435 def __len__(self) -> int: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
436 return len(str(self._url)) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
438 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
439 def build( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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.
454 This method takes either `hosts` - a list of `MultiHostHost` typed dicts, or the individual components
455 `username`, `password`, `host` and `port`.
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.
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 )
485 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
486 def serialize_url(cls, url: Any, info: core_schema.SerializationInfo) -> str | Self: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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
495 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
496 def __get_pydantic_core_schema__( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
497 cls, source: type[_BaseMultiHostUrl], handler: GetCoreSchemaHandler
498 ) -> core_schema.CoreSchema:
499 def wrap_val(v, h): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
500 if isinstance(v, source): 500 ↛ 501line 500 didn't jump to line 501 because the condition on line 500 was never true1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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 true1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
503 v = str(v)
504 core_url = h(v) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
505 instance = source.__new__(source) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
506 instance._url = core_url 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
507 return instance 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
509 return core_schema.no_info_wrap_validator_function( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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 )
517 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
518 def __get_pydantic_json_schema__( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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)
526 __pydantic_serializer__ = SchemaSerializer(core_schema.any_schema(serialization=core_schema.to_string_ser_schema())) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
529@lru_cache 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
530def _build_type_adapter(cls: type[_BaseUrl | _BaseMultiHostUrl]) -> TypeAdapter: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
531 return TypeAdapter(cls) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
534class AnyUrl(_BaseUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
535 """Base type for all URLs.
537 * Any scheme allowed
538 * Top-level domain (TLD) not required
539 * Host not required
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:
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 """
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`
559class AnyHttpUrl(AnyUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
560 """A type that will accept any http or https URL.
562 * TLD not required
563 * Host not required
564 """
566 _constraints = UrlConstraints(allowed_schemes=['http', 'https']) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
569class HttpUrl(AnyUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
570 """A type that will accept any http or https URL.
572 * TLD not required
573 * Host not required
574 * Max length 2083
576 ```python
577 from pydantic import BaseModel, HttpUrl, ValidationError
579 class MyModel(BaseModel):
580 url: HttpUrl
582 m = MyModel(url='http://www.example.com') # (1)!
583 print(m.url)
584 #> http://www.example.com/
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 '''
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 ```
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.
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):
613 ```python
614 from pydantic import BaseModel, HttpUrl
616 class MyModel(BaseModel):
617 url: HttpUrl
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 ```
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.
635 To explain this; consider the following two cases:
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.
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.
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 """
647 _constraints = UrlConstraints(max_length=2083, allowed_schemes=['http', 'https']) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
650class AnyWebsocketUrl(AnyUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
651 """A type that will accept any ws or wss URL.
653 * TLD not required
654 * Host not required
655 """
657 _constraints = UrlConstraints(allowed_schemes=['ws', 'wss']) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
660class WebsocketUrl(AnyUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
661 """A type that will accept any ws or wss URL.
663 * TLD not required
664 * Host not required
665 * Max length 2083
666 """
668 _constraints = UrlConstraints(max_length=2083, allowed_schemes=['ws', 'wss']) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
671class FileUrl(AnyUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
672 """A type that will accept any file URL.
674 * Host not required
675 """
677 _constraints = UrlConstraints(allowed_schemes=['file']) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
680class FtpUrl(AnyUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
681 """A type that will accept ftp URL.
683 * TLD not required
684 * Host not required
685 """
687 _constraints = UrlConstraints(allowed_schemes=['ftp']) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
690class PostgresDsn(_BaseMultiHostUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
691 """A type that will accept any Postgres DSN.
693 * User info required
694 * TLD not required
695 * Host required
696 * Supports multiple hosts
698 If further validation is required, these properties can be used by validators to enforce specific behaviour:
700 ```python
701 from pydantic import (
702 BaseModel,
703 HttpUrl,
704 PostgresDsn,
705 ValidationError,
706 field_validator,
707 )
709 class MyModel(BaseModel):
710 url: HttpUrl
712 m = MyModel(url='http://www.example.com')
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
724 class MyDatabaseModel(BaseModel):
725 db: PostgresDsn
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
732 m = MyDatabaseModel(db='postgres://user:pass@localhost:5432/foobar')
733 print(m.db)
734 #> postgres://user:pass@localhost:5432/foobar
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 """
750 _constraints = UrlConstraints( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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 )
765 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
766 def host(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
767 """The required URL host."""
768 return self._url.host # pyright: ignore[reportAttributeAccessIssue]
771class CockroachDsn(AnyUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
772 """A type that will accept any Cockroach DSN.
774 * User info required
775 * TLD not required
776 * Host required
777 """
779 _constraints = UrlConstraints( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
780 host_required=True,
781 allowed_schemes=[
782 'cockroachdb',
783 'cockroachdb+psycopg2',
784 'cockroachdb+asyncpg',
785 ],
786 )
788 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
789 def host(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
790 """The required URL host."""
791 return self._url.host # pyright: ignore[reportReturnType]
794class AmqpDsn(AnyUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
795 """A type that will accept any AMQP DSN.
797 * User info required
798 * TLD not required
799 * Host not required
800 """
802 _constraints = UrlConstraints(allowed_schemes=['amqp', 'amqps']) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
805class RedisDsn(AnyUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
806 """A type that will accept any Redis DSN.
808 * User info required
809 * TLD not required
810 * Host required (e.g., `rediss://:pass@localhost`)
811 """
813 _constraints = UrlConstraints( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
814 allowed_schemes=['redis', 'rediss'],
815 default_host='localhost',
816 default_port=6379,
817 default_path='/0',
818 host_required=True,
819 )
821 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
822 def host(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
823 """The required URL host."""
824 return self._url.host # pyright: ignore[reportReturnType] 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
827class MongoDsn(_BaseMultiHostUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
828 """A type that will accept any MongoDB DSN.
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 """
836 _constraints = UrlConstraints(allowed_schemes=['mongodb', 'mongodb+srv'], default_port=27017) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
839class KafkaDsn(AnyUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
840 """A type that will accept any Kafka DSN.
842 * User info required
843 * TLD not required
844 * Host not required
845 """
847 _constraints = UrlConstraints(allowed_schemes=['kafka'], default_host='localhost', default_port=9092) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
850class NatsDsn(_BaseMultiHostUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
851 """A type that will accept any NATS DSN.
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 """
859 _constraints = UrlConstraints( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
860 allowed_schemes=['nats', 'tls', 'ws', 'wss'], default_host='localhost', default_port=4222
861 )
864class MySQLDsn(AnyUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
865 """A type that will accept any MySQL DSN.
867 * User info required
868 * TLD not required
869 * Host not required
870 """
872 _constraints = UrlConstraints( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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 )
888class MariaDBDsn(AnyUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
889 """A type that will accept any MariaDB DSN.
891 * User info required
892 * TLD not required
893 * Host not required
894 """
896 _constraints = UrlConstraints( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
897 allowed_schemes=['mariadb', 'mariadb+mariadbconnector', 'mariadb+pymysql'],
898 default_port=3306,
899 )
902class ClickHouseDsn(AnyUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
903 """A type that will accept any ClickHouse DSN.
905 * User info required
906 * TLD not required
907 * Host not required
908 """
910 _constraints = UrlConstraints( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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 )
924class SnowflakeDsn(AnyUrl): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
925 """A type that will accept any Snowflake DSN.
927 * User info required
928 * TLD not required
929 * Host required
930 """
932 _constraints = UrlConstraints( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
933 allowed_schemes=['snowflake'],
934 host_required=True,
935 )
937 @property 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
938 def host(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
939 """The required URL host."""
940 return self._url.host # pyright: ignore[reportReturnType]
943def import_email_validator() -> None: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
944 global email_validator
945 try: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
946 import email_validator 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
947 except ImportError as e: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
948 raise ImportError("email-validator is not installed, run `pip install 'pydantic[email]'`") from e 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
949 if not version('email-validator').partition('.')[0] == '2': 1nbcdeoapfghiqjklm
950 raise ImportError('email-validator version >= 2.0 required, run pip install -U email-validator') 1nbcdeoapfghiqjklm
953if TYPE_CHECKING: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
954 EmailStr = Annotated[str, ...]
955else:
957 class EmailStr: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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:
963 ```bash
964 pip install email-validator
965 ```
967 Validate email addresses.
969 ```python
970 from pydantic import BaseModel, EmailStr
972 class Model(BaseModel):
973 email: EmailStr
975 print(Model(email='contact@mail.com'))
976 #> email='contact@mail.com'
977 ```
978 """ # noqa: D212
980 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
981 def __get_pydantic_core_schema__( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
982 cls,
983 _source: type[Any],
984 _handler: GetCoreSchemaHandler,
985 ) -> core_schema.CoreSchema:
986 import_email_validator() 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
987 return core_schema.no_info_after_validator_function(cls._validate, core_schema.str_schema()) 1nbcdeoapfghiqjklm
989 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
990 def __get_pydantic_json_schema__( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
991 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler
992 ) -> JsonSchemaValue:
993 field_schema = handler(core_schema) 1nbcdeoapfghiqjklm
994 field_schema.update(type='string', format='email') 1nbcdeoapfghiqjklm
995 return field_schema 1nbcdeoapfghiqjklm
997 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
998 def _validate(cls, input_value: str, /) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
999 return validate_email(input_value)[1] 1nbcdeoapfghiqjklm
1002class NameEmail(_repr.Representation): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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:
1008 ```bash
1009 pip install email-validator
1010 ```
1012 Validate a name and email address combination, as specified by
1013 [RFC 5322](https://datatracker.ietf.org/doc/html/rfc5322#section-3.4).
1015 The `NameEmail` has two properties: `name` and `email`.
1016 In case the `name` is not provided, it's inferred from the email address.
1018 ```python
1019 from pydantic import BaseModel, NameEmail
1021 class User(BaseModel):
1022 email: NameEmail
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
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
1038 __slots__ = 'name', 'email' 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1040 def __init__(self, name: str, email: str): 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1041 self.name = name 1nbcdeoapfghiqjklm
1042 self.email = email 1nbcdeoapfghiqjklm
1044 def __eq__(self, other: Any) -> bool: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1045 return isinstance(other, NameEmail) and (self.name, self.email) == (other.name, other.email) 1nbcdeoapfghiqjklm
1047 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1048 def __get_pydantic_json_schema__( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1049 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler
1050 ) -> JsonSchemaValue:
1051 field_schema = handler(core_schema) 1nbcdeoapfghiqjklm
1052 field_schema.update(type='string', format='name-email') 1nbcdeoapfghiqjklm
1053 return field_schema 1nbcdeoapfghiqjklm
1055 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1056 def __get_pydantic_core_schema__( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1057 cls,
1058 _source: type[Any],
1059 _handler: GetCoreSchemaHandler,
1060 ) -> core_schema.CoreSchema:
1061 import_email_validator() 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1063 return core_schema.no_info_after_validator_function( 1nbcdeoapfghiqjklm
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 )
1076 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1077 def _validate(cls, input_value: Self | str, /) -> Self: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1078 if isinstance(input_value, str): 1nbcdeoapfghiqjklm
1079 name, email = validate_email(input_value) 1nbcdeoapfghiqjklm
1080 return cls(name, email) 1nbcdeoapfghiqjklm
1081 else:
1082 return input_value 1nbcdeoapfghiqjklm
1084 def __str__(self) -> str: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1085 if '@' in self.name: 1nbcdeoapfghiqjklm
1086 return f'"{self.name}" <{self.email}>' 1nbcdeoapfghiqjklm
1088 return f'{self.name} <{self.email}>' 1nbcdeoapfghiqjklm
1091IPvAnyAddressType: TypeAlias = 'IPv4Address | IPv6Address' 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1092IPvAnyInterfaceType: TypeAlias = 'IPv4Interface | IPv6Interface' 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1093IPvAnyNetworkType: TypeAlias = 'IPv4Network | IPv6Network' 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1095if TYPE_CHECKING: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1096 IPvAnyAddress = IPvAnyAddressType
1097 IPvAnyInterface = IPvAnyInterfaceType
1098 IPvAnyNetwork = IPvAnyNetworkType
1099else:
1101 class IPvAnyAddress: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1102 """Validate an IPv4 or IPv6 address.
1104 ```python
1105 from pydantic import BaseModel
1106 from pydantic.networks import IPvAnyAddress
1108 class IpModel(BaseModel):
1109 ip: IPvAnyAddress
1111 print(IpModel(ip='127.0.0.1'))
1112 #> ip=IPv4Address('127.0.0.1')
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 """
1131 __slots__ = () 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1133 def __new__(cls, value: Any) -> IPvAnyAddressType: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1134 """Validate an IPv4 or IPv6 address."""
1135 try: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1136 return IPv4Address(value) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1137 except ValueError: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1138 pass 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1140 try: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1141 return IPv6Address(value) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1142 except ValueError: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1143 raise PydanticCustomError('ip_any_address', 'value is not a valid IPv4 or IPv6 address') 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1145 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1146 def __get_pydantic_json_schema__( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1147 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler
1148 ) -> JsonSchemaValue:
1149 field_schema = {} 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1150 field_schema.update(type='string', format='ipvanyaddress') 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1151 return field_schema 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1153 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1154 def __get_pydantic_core_schema__( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1155 cls,
1156 _source: type[Any],
1157 _handler: GetCoreSchemaHandler,
1158 ) -> core_schema.CoreSchema:
1159 return core_schema.no_info_plain_validator_function( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1160 cls._validate, serialization=core_schema.to_string_ser_schema()
1161 )
1163 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1164 def _validate(cls, input_value: Any, /) -> IPvAnyAddressType: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1165 return cls(input_value) # type: ignore[return-value] 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1167 class IPvAnyInterface: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1168 """Validate an IPv4 or IPv6 interface."""
1170 __slots__ = () 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1172 def __new__(cls, value: NetworkType) -> IPvAnyInterfaceType: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1173 """Validate an IPv4 or IPv6 interface."""
1174 try: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1175 return IPv4Interface(value) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1176 except ValueError: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1177 pass 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1179 try: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1180 return IPv6Interface(value) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1181 except ValueError: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1182 raise PydanticCustomError('ip_any_interface', 'value is not a valid IPv4 or IPv6 interface') 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1184 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1185 def __get_pydantic_json_schema__( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1186 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler
1187 ) -> JsonSchemaValue:
1188 field_schema = {} 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1189 field_schema.update(type='string', format='ipvanyinterface') 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1190 return field_schema 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1192 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1193 def __get_pydantic_core_schema__( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1194 cls,
1195 _source: type[Any],
1196 _handler: GetCoreSchemaHandler,
1197 ) -> core_schema.CoreSchema:
1198 return core_schema.no_info_plain_validator_function( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1199 cls._validate, serialization=core_schema.to_string_ser_schema()
1200 )
1202 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1203 def _validate(cls, input_value: NetworkType, /) -> IPvAnyInterfaceType: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1204 return cls(input_value) # type: ignore[return-value] 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1206 class IPvAnyNetwork: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1207 """Validate an IPv4 or IPv6 network."""
1209 __slots__ = () 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1211 def __new__(cls, value: NetworkType) -> IPvAnyNetworkType: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
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: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1216 return IPv4Network(value) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1217 except ValueError: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1218 pass 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1220 try: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1221 return IPv6Network(value) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1222 except ValueError: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1223 raise PydanticCustomError('ip_any_network', 'value is not a valid IPv4 or IPv6 network') 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1225 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1226 def __get_pydantic_json_schema__( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1227 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler
1228 ) -> JsonSchemaValue:
1229 field_schema = {} 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1230 field_schema.update(type='string', format='ipvanynetwork') 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1231 return field_schema 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1233 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1234 def __get_pydantic_core_schema__( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1235 cls,
1236 _source: type[Any],
1237 _handler: GetCoreSchemaHandler,
1238 ) -> core_schema.CoreSchema:
1239 return core_schema.no_info_plain_validator_function( 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1240 cls._validate, serialization=core_schema.to_string_ser_schema()
1241 )
1243 @classmethod 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1244 def _validate(cls, input_value: NetworkType, /) -> IPvAnyNetworkType: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1245 return cls(input_value) # type: ignore[return-value] 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1248def _build_pretty_email_regex() -> re.Pattern[str]: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1249 name_chars = r'[\w!#$%&\'*+\-/=?^_`{|}~]' 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1250 unquoted_name_group = rf'((?:{name_chars}+\s+)*{name_chars}+)' 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1251 quoted_name_group = r'"((?:[^"]|\")+)"' 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1252 email_group = r'<(.+)>' 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1253 return re.compile(rf'\s*(?:{unquoted_name_group}|{quoted_name_group})?\s*{email_group}\s*') 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1256pretty_email_regex = _build_pretty_email_regex() 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1258MAX_EMAIL_LENGTH = 2048 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1259"""Maximum length for an email. 1brcsdteuvafwgxhyizAjBkClDmEF
1260A somewhat arbitrary but very generous number compared to what is allowed by most implementations.
1261"""
1264def validate_email(value: str) -> tuple[str, str]: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1265 """Email address validation using [email-validator](https://pypi.org/project/email-validator/).
1267 Returns:
1268 A tuple containing the local part of the email (or the name for "pretty" email addresses)
1269 and the normalized email.
1271 Raises:
1272 PydanticCustomError: If the email is invalid.
1274 Note:
1275 Note that:
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: 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1282 import_email_validator() 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF
1284 if len(value) > MAX_EMAIL_LENGTH: 1nbcdeoapfghiqjklm
1285 raise PydanticCustomError( 1nbcdeoapfghiqjklm
1286 'value_error',
1287 'value is not a valid email address: {reason}',
1288 {'reason': f'Length must not exceed {MAX_EMAIL_LENGTH} characters'},
1289 )
1291 m = pretty_email_regex.fullmatch(value) 1nbcdeoapfghiqjklm
1292 name: str | None = None 1nbcdeoapfghiqjklm
1293 if m: 1nbcdeoapfghiqjklm
1294 unquoted_name, quoted_name, value = m.groups() 1nbcdeoapfghiqjklm
1295 name = unquoted_name or quoted_name 1nbcdeoapfghiqjklm
1297 email = value.strip() 1nbcdeoapfghiqjklm
1299 try: 1nbcdeoapfghiqjklm
1300 parts = email_validator.validate_email(email, check_deliverability=False) 1nbcdeoapfghiqjklm
1301 except email_validator.EmailNotValidError as e: 1nbcdeoapfghiqjklm
1302 raise PydanticCustomError( 1nbcdeoapfghiqjklm
1303 'value_error', 'value is not a valid email address: {reason}', {'reason': str(e.args[0])}
1304 ) from e
1306 email = parts.normalized 1nbcdeoapfghiqjklm
1307 assert email is not None 1nbcdeoapfghiqjklm
1308 name = name or parts.local_part 1nbcdeoapfghiqjklm
1309 return name, email 1nbcdeoapfghiqjklm
1312__getattr__ = getattr_migration(__name__) 1nGbrcsdteuvoapHfwgxhyizAqIjBkClDmEF