Coverage for pydantic/networks.py: 92.29%
383 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-22 09:30 +0000
1"""The networks module contains types for common network-related fields."""
3from __future__ import annotations as _annotations 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
5import dataclasses as _dataclasses 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
6import re 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
7from dataclasses import fields 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
8from functools import lru_cache 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
9from importlib.metadata import version 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
10from ipaddress import IPv4Address, IPv4Interface, IPv4Network, IPv6Address, IPv6Interface, IPv6Network 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
11from typing import TYPE_CHECKING, Annotated, Any, ClassVar 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
13from pydantic_core import ( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
14 MultiHostHost,
15 PydanticCustomError,
16 PydanticSerializationUnexpectedValue,
17 SchemaSerializer,
18 core_schema,
19)
20from pydantic_core import MultiHostUrl as _CoreMultiHostUrl 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
21from pydantic_core import Url as _CoreUrl 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
22from typing_extensions import Self, TypeAlias 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
24from pydantic.errors import PydanticUserError 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
26from ._internal import _repr, _schema_generation_shared 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
27from ._migration import getattr_migration 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
28from .annotated_handlers import GetCoreSchemaHandler 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
29from .json_schema import JsonSchemaValue 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
30from .type_adapter import TypeAdapter 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
32if TYPE_CHECKING: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
33 import email_validator
35 NetworkType: TypeAlias = 'str | bytes | int | tuple[str | bytes | int, str | int]' 1a
37else:
38 email_validator = None 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
41__all__ = [ 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
71class UrlConstraints: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
84 allowed_schemes: list[str] | None = None 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
85 host_required: bool | None = None 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
86 default_host: str | None = None 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
87 default_port: int | None = None 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
88 default_path: str | None = None 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
90 def __hash__(self) -> int: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
91 return hash( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
103 def defined_constraints(self) -> dict[str, Any]: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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} 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
107 def __get_pydantic_core_schema__(self, source: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
108 schema = handler(source) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 true1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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(): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
119 schema_to_mutate[constraint_key] = constraint_value 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
120 return schema 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
123class _BaseUrl: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
124 _constraints: ClassVar[UrlConstraints] = UrlConstraints() 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
125 _url: _CoreUrl 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
127 def __init__(self, url: str | _CoreUrl | _BaseUrl) -> None: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
128 self._url = _build_type_adapter(self.__class__).validate_python(url)._url 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
130 @property 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
131 def scheme(self) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
138 @property 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
139 def username(self) -> str | None: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
146 @property 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
147 def password(self) -> str | None: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
154 @property 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
155 def host(self) -> str | None: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
163 def unicode_host(self) -> str | None: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
174 def port(self) -> int | None: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
181 @property 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
182 def path(self) -> str | None: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
189 @property 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
190 def query(self) -> str | None: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
197 def query_params(self) -> list[tuple[str, str]]: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
205 def fragment(self) -> str | None: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
212 def unicode_string(self) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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() 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
220 def encoded_string(self) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
227 def __str__(self) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
228 """The URL as a string, this will punycode encode the host if required."""
229 return str(self._url) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
231 def __repr__(self) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
232 return f'{self.__class__.__name__}({str(self._url)!r})' 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
234 def __deepcopy__(self, memo: dict) -> Self: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
235 return self.__class__(self._url)
237 def __eq__(self, other: Any) -> bool: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
238 return self.__class__ is other.__class__ and self._url == other._url 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
240 def __lt__(self, other: Any) -> bool: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
241 return self.__class__ is other.__class__ and self._url < other._url 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
243 def __gt__(self, other: Any) -> bool: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
244 return self.__class__ is other.__class__ and self._url > other._url 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
246 def __le__(self, other: Any) -> bool: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
247 return self.__class__ is other.__class__ and self._url <= other._url 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
249 def __ge__(self, other: Any) -> bool: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
250 return self.__class__ is other.__class__ and self._url >= other._url 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
252 def __hash__(self) -> int: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
253 return hash(self._url) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
255 def __len__(self) -> int: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
256 return len(str(self._url)) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
258 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
259 def build( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
300 def serialize_url(cls, url: Any, info: core_schema.SerializationInfo) -> str | Self: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
301 if not isinstance(url, cls): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
302 raise PydanticSerializationUnexpectedValue( 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
303 f"Expected `{cls}` but got `{type(url)}` with value `'{url}'` - serialized value may not be as expected."
304 )
305 if info.mode == 'json': 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
306 return str(url) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
307 return url 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
309 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
310 def __get_pydantic_core_schema__( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
311 cls, source: type[_BaseUrl], handler: GetCoreSchemaHandler
312 ) -> core_schema.CoreSchema:
313 def wrap_val(v, h): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
314 if isinstance(v, source): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
315 return v 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
316 if isinstance(v, _BaseUrl): 316 ↛ 317line 316 didn't jump to line 317 because the condition on line 316 was never true1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
317 v = str(v)
318 core_url = h(v) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
319 instance = source.__new__(source) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
320 instance._url = core_url 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
321 return instance 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
323 return core_schema.no_info_wrap_validator_function( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
332 def __get_pydantic_json_schema__( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
338 return handler(inner_schema) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
340 __pydantic_serializer__ = SchemaSerializer(core_schema.any_schema(serialization=core_schema.to_string_ser_schema())) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
343class _BaseMultiHostUrl: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
344 _constraints: ClassVar[UrlConstraints] = UrlConstraints() 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
345 _url: _CoreMultiHostUrl 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
347 def __init__(self, url: str | _CoreMultiHostUrl | _BaseMultiHostUrl) -> None: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
348 self._url = _build_type_adapter(self.__class__).validate_python(url)._url 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
350 @property 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
351 def scheme(self) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
358 @property 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
359 def path(self) -> str | None: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
366 @property 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
367 def query(self) -> str | None: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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]]: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
382 def fragment(self) -> str | None: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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]: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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() 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
408 def encoded_string(self) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
415 def unicode_string(self) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
420 """The URL as a string, this will punycode encode the host if required."""
421 return str(self._url) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
423 def __repr__(self) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
424 return f'{self.__class__.__name__}({str(self._url)!r})' 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
426 def __deepcopy__(self, memo: dict) -> Self: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
427 return self.__class__(self._url)
429 def __eq__(self, other: Any) -> bool: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
430 return self.__class__ is other.__class__ and self._url == other._url 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
432 def __hash__(self) -> int: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
433 return hash(self._url) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
435 def __len__(self) -> int: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
436 return len(str(self._url)) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
438 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
439 def build( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
486 def serialize_url(cls, url: Any, info: core_schema.SerializationInfo) -> str | Self: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
496 def __get_pydantic_core_schema__( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
497 cls, source: type[_BaseMultiHostUrl], handler: GetCoreSchemaHandler
498 ) -> core_schema.CoreSchema:
499 def wrap_val(v, h): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
500 if isinstance(v, source): 500 ↛ 501line 500 didn't jump to line 501 because the condition on line 500 was never true1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 true1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
503 v = str(v)
504 core_url = h(v) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
505 instance = source.__new__(source) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
506 instance._url = core_url 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
507 return instance 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
509 return core_schema.no_info_wrap_validator_function( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
518 def __get_pydantic_json_schema__( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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())) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
529@lru_cache 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
530def _build_type_adapter(cls: type[_BaseUrl | _BaseMultiHostUrl]) -> TypeAdapter: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
531 return TypeAdapter(cls) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
534class AnyUrl(_BaseUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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']) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
569class HttpUrl(AnyUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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']) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
650class AnyWebsocketUrl(AnyUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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']) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
660class WebsocketUrl(AnyUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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']) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
671class FileUrl(AnyUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
672 """A type that will accept any file URL.
674 * Host not required
675 """
677 _constraints = UrlConstraints(allowed_schemes=['file']) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
680class FtpUrl(AnyUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
681 """A type that will accept ftp URL.
683 * TLD not required
684 * Host not required
685 """
687 _constraints = UrlConstraints(allowed_schemes=['ftp']) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
690class PostgresDsn(_BaseMultiHostUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
766 def host(self) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
767 """The required URL host."""
768 return self._url.host # pyright: ignore[reportAttributeAccessIssue]
771class CockroachDsn(AnyUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
772 """A type that will accept any Cockroach DSN.
774 * User info required
775 * TLD not required
776 * Host required
777 """
779 _constraints = UrlConstraints( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
780 host_required=True,
781 allowed_schemes=[
782 'cockroachdb',
783 'cockroachdb+psycopg2',
784 'cockroachdb+asyncpg',
785 ],
786 )
788 @property 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
789 def host(self) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
790 """The required URL host."""
791 return self._url.host # pyright: ignore[reportReturnType]
794class AmqpDsn(AnyUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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']) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
805class RedisDsn(AnyUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
814 allowed_schemes=['redis', 'rediss'],
815 default_host='localhost',
816 default_port=6379,
817 default_path='/0',
818 host_required=True,
819 )
821 @property 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
822 def host(self) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
823 """The required URL host."""
824 return self._url.host # pyright: ignore[reportReturnType] 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
827class MongoDsn(_BaseMultiHostUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
839class KafkaDsn(AnyUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
850class NatsDsn(_BaseMultiHostUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
860 allowed_schemes=['nats', 'tls', 'ws', 'wss'], default_host='localhost', default_port=4222
861 )
864class MySQLDsn(AnyUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
897 allowed_schemes=['mariadb', 'mariadb+mariadbconnector', 'mariadb+pymysql'],
898 default_port=3306,
899 )
902class ClickHouseDsn(AnyUrl): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
925 """A type that will accept any Snowflake DSN.
927 * User info required
928 * TLD not required
929 * Host required
930 """
932 _constraints = UrlConstraints( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
933 allowed_schemes=['snowflake'],
934 host_required=True,
935 )
937 @property 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
938 def host(self) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
939 """The required URL host."""
940 return self._url.host # pyright: ignore[reportReturnType]
943def import_email_validator() -> None: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
944 global email_validator
945 try: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
946 import email_validator 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
947 except ImportError as e: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
948 raise ImportError("email-validator is not installed, run `pip install 'pydantic[email]'`") from e 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
949 if not version('email-validator').partition('.')[0] == '2': 1qbcdefrasghijktlmnop
950 raise ImportError('email-validator version >= 2.0 required, run pip install -U email-validator') 1qbcdefrasghijktlmnop
953if TYPE_CHECKING: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
954 EmailStr = Annotated[str, ...]
955else:
957 class EmailStr: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
981 def __get_pydantic_core_schema__( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
982 cls,
983 _source: type[Any],
984 _handler: GetCoreSchemaHandler,
985 ) -> core_schema.CoreSchema:
986 import_email_validator() 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
987 return core_schema.no_info_after_validator_function(cls._validate, core_schema.str_schema()) 1qbcdefrasghijktlmnop
989 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
990 def __get_pydantic_json_schema__( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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
997 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
998 def _validate(cls, input_value: str, /) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
999 return validate_email(input_value)[1] 1qbcdefrasghijktlmnop
1002class NameEmail(_repr.Representation): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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' 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1040 def __init__(self, name: str, email: str): 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1041 self.name = name 1qbcdefrasghijktlmnop
1042 self.email = email 1qbcdefrasghijktlmnop
1044 def __eq__(self, other: Any) -> bool: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1045 return isinstance(other, NameEmail) and (self.name, self.email) == (other.name, other.email) 1qbcdefrasghijktlmnop
1047 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1048 def __get_pydantic_json_schema__( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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
1055 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1056 def __get_pydantic_core_schema__( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1057 cls,
1058 _source: type[Any],
1059 _handler: GetCoreSchemaHandler,
1060 ) -> core_schema.CoreSchema:
1061 import_email_validator() 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 )
1076 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1077 def _validate(cls, input_value: Self | str, /) -> Self: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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
1084 def __str__(self) -> str: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1085 if '@' in self.name: 1qbcdefrasghijktlmnop
1086 return f'"{self.name}" <{self.email}>' 1qbcdefrasghijktlmnop
1088 return f'{self.name} <{self.email}>' 1qbcdefrasghijktlmnop
1091IPvAnyAddressType: TypeAlias = 'IPv4Address | IPv6Address' 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1092IPvAnyInterfaceType: TypeAlias = 'IPv4Interface | IPv6Interface' 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1093IPvAnyNetworkType: TypeAlias = 'IPv4Network | IPv6Network' 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1095if TYPE_CHECKING: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1096 IPvAnyAddress = IPvAnyAddressType
1097 IPvAnyInterface = IPvAnyInterfaceType
1098 IPvAnyNetwork = IPvAnyNetworkType
1099else:
1101 class IPvAnyAddress: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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__ = () 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1133 def __new__(cls, value: Any) -> IPvAnyAddressType: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1134 """Validate an IPv4 or IPv6 address."""
1135 try: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1136 return IPv4Address(value) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1137 except ValueError: 1qJbucvdwexfyrasKgzhAiBjCkDMtLlEmFnGoHpIN
1138 pass 1qJbucvdwexfyrasKgzhAiBjCkDMtLlEmFnGoHpIN
1140 try: 1qJbucvdwexfyrasKgzhAiBjCkDMtLlEmFnGoHpIN
1141 return IPv6Address(value) 1qJbucvdwexfyrasKgzhAiBjCkDMtLlEmFnGoHpIN
1142 except ValueError: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
1143 raise PydanticCustomError('ip_any_address', 'value is not a valid IPv4 or IPv6 address') 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
1145 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1146 def __get_pydantic_json_schema__( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1147 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler
1148 ) -> JsonSchemaValue:
1149 field_schema = {} 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1150 field_schema.update(type='string', format='ipvanyaddress') 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1151 return field_schema 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1153 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1154 def __get_pydantic_core_schema__( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1155 cls,
1156 _source: type[Any],
1157 _handler: GetCoreSchemaHandler,
1158 ) -> core_schema.CoreSchema:
1159 return core_schema.no_info_plain_validator_function( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1160 cls._validate, serialization=core_schema.to_string_ser_schema()
1161 )
1163 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1164 def _validate(cls, input_value: Any, /) -> IPvAnyAddressType: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1165 return cls(input_value) # type: ignore[return-value] 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1167 class IPvAnyInterface: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1168 """Validate an IPv4 or IPv6 interface."""
1170 __slots__ = () 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1172 def __new__(cls, value: NetworkType) -> IPvAnyInterfaceType: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1173 """Validate an IPv4 or IPv6 interface."""
1174 try: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1175 return IPv4Interface(value) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1176 except ValueError: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
1177 pass 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
1179 try: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
1180 return IPv6Interface(value) 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
1181 except ValueError: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
1182 raise PydanticCustomError('ip_any_interface', 'value is not a valid IPv4 or IPv6 interface') 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
1184 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1185 def __get_pydantic_json_schema__( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1186 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler
1187 ) -> JsonSchemaValue:
1188 field_schema = {} 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1189 field_schema.update(type='string', format='ipvanyinterface') 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1190 return field_schema 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1192 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1193 def __get_pydantic_core_schema__( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1194 cls,
1195 _source: type[Any],
1196 _handler: GetCoreSchemaHandler,
1197 ) -> core_schema.CoreSchema:
1198 return core_schema.no_info_plain_validator_function( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1199 cls._validate, serialization=core_schema.to_string_ser_schema()
1200 )
1202 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1203 def _validate(cls, input_value: NetworkType, /) -> IPvAnyInterfaceType: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1204 return cls(input_value) # type: ignore[return-value] 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1206 class IPvAnyNetwork: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1207 """Validate an IPv4 or IPv6 network."""
1209 __slots__ = () 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1211 def __new__(cls, value: NetworkType) -> IPvAnyNetworkType: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1216 return IPv4Network(value) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1217 except ValueError: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
1218 pass 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
1220 try: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
1221 return IPv6Network(value) 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
1222 except ValueError: 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
1223 raise PydanticCustomError('ip_any_network', 'value is not a valid IPv4 or IPv6 network') 1qJbucvdwexfyrasKgzhAiBjCkDtLlEmFnGoHpI
1225 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1226 def __get_pydantic_json_schema__( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1227 cls, core_schema: core_schema.CoreSchema, handler: _schema_generation_shared.GetJsonSchemaHandler
1228 ) -> JsonSchemaValue:
1229 field_schema = {} 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1230 field_schema.update(type='string', format='ipvanynetwork') 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1231 return field_schema 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1233 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1234 def __get_pydantic_core_schema__( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1235 cls,
1236 _source: type[Any],
1237 _handler: GetCoreSchemaHandler,
1238 ) -> core_schema.CoreSchema:
1239 return core_schema.no_info_plain_validator_function( 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1240 cls._validate, serialization=core_schema.to_string_ser_schema()
1241 )
1243 @classmethod 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1244 def _validate(cls, input_value: NetworkType, /) -> IPvAnyNetworkType: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1245 return cls(input_value) # type: ignore[return-value] 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1248def _build_pretty_email_regex() -> re.Pattern[str]: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1249 name_chars = r'[\w!#$%&\'*+\-/=?^_`{|}~]' 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1250 unquoted_name_group = rf'((?:{name_chars}+\s+)*{name_chars}+)' 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1251 quoted_name_group = r'"((?:[^"]|\")+)"' 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1252 email_group = r'<(.+)>' 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1253 return re.compile(rf'\s*(?:{unquoted_name_group}|{quoted_name_group})?\s*{email_group}\s*') 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1256pretty_email_regex = _build_pretty_email_regex() 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1258MAX_EMAIL_LENGTH = 2048 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1259"""Maximum length for an email. 1bucvdwexfyOagzhAiBjCkDMlEmFnGoHpIN
1260A somewhat arbitrary but very generous number compared to what is allowed by most implementations.
1261"""
1264def validate_email(value: str) -> tuple[str, str]: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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: 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
1282 import_email_validator() 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN
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 )
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
1297 email = value.strip() 1qbcdefrasghijktlmnop
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
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
1312__getattr__ = getattr_migration(__name__) 1qJbucvdwexfyOrasKgzhAiBjCkDMtLlEmFnGoHpIN