Coverage for pydantic/networks.py: 97.54%

200 statements  

« prev     ^ index     » next       coverage.py v7.5.3, created at 2024-06-21 17:00 +0000

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

2 

3from __future__ import annotations as _annotations 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

4 

5import dataclasses as _dataclasses 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

6import re 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

7from importlib.metadata import version 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

8from ipaddress import IPv4Address, IPv4Interface, IPv4Network, IPv6Address, IPv6Interface, IPv6Network 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

9from typing import TYPE_CHECKING, Any 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

10 

11from pydantic_core import MultiHostUrl, PydanticCustomError, Url, core_schema 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

12from typing_extensions import Annotated, Self, TypeAlias 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

13 

14from ._internal import _fields, _repr, _schema_generation_shared 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

15from ._migration import getattr_migration 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

16from .annotated_handlers import GetCoreSchemaHandler 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

17from .json_schema import JsonSchemaValue 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

18 

19if TYPE_CHECKING: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

20 import email_validator 

21 

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

23 

24else: 

25 email_validator = None 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

26 

27 

28__all__ = [ 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

29 'AnyUrl', 

30 'AnyHttpUrl', 

31 'FileUrl', 

32 'FtpUrl', 

33 'HttpUrl', 

34 'WebsocketUrl', 

35 'AnyWebsocketUrl', 

36 'UrlConstraints', 

37 'EmailStr', 

38 'NameEmail', 

39 'IPvAnyAddress', 

40 'IPvAnyInterface', 

41 'IPvAnyNetwork', 

42 'PostgresDsn', 

43 'CockroachDsn', 

44 'AmqpDsn', 

45 'RedisDsn', 

46 'MongoDsn', 

47 'KafkaDsn', 

48 'NatsDsn', 

49 'validate_email', 

50 'MySQLDsn', 

51 'MariaDBDsn', 

52 'ClickHouseDsn', 

53] 

54 

55 

56@_dataclasses.dataclass 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

57class UrlConstraints(_fields.PydanticMetadata): 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

58 """Url constraints. 

59 

60 Attributes: 

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

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

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

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

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

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

67 """ 

68 

69 max_length: int | None = None 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

70 allowed_schemes: list[str] | None = None 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

71 host_required: bool | None = None 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

72 default_host: str | None = None 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

73 default_port: int | None = None 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

74 default_path: str | None = None 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

75 

76 def __hash__(self) -> int: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

77 return hash( 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

78 ( 

79 self.max_length, 

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

81 self.host_required, 

82 self.default_host, 

83 self.default_port, 

84 self.default_path, 

85 ) 

86 ) 

87 

88 

89AnyUrl = Url 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

90"""Base type for all URLs. 1bcdefgahijklmGHIJKLMnopqrs

91 

92* Any scheme allowed 

93* Top-level domain (TLD) not required 

94* Host required 

95 

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

97the types export the following properties: 

98 

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

100- `host`: the URL host (`example.com`), always set. 

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

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

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

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

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

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

107""" 

108AnyHttpUrl = Annotated[Url, UrlConstraints(allowed_schemes=['http', 'https'])] 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

109"""A type that will accept any http or https URL. 1bcdefgahijklmGHIJKLMnopqrs

110 

111* TLD not required 

112* Host required 

113""" 

114HttpUrl = Annotated[Url, UrlConstraints(max_length=2083, allowed_schemes=['http', 'https'])] 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

115"""A type that will accept any http or https URL. 1bcdefgahijklmGHIJKLMnopqrs

116 

117* TLD not required 

118* Host required 

119* Max length 2083 

120 

121```py 

122from pydantic import BaseModel, HttpUrl, ValidationError 

123 

124class MyModel(BaseModel): 

125 url: HttpUrl 

126 

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

128print(m.url) 

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

130 

131try: 

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

133except ValidationError as e: 

134 print(e) 

135 ''' 

136 1 validation error for MyModel 

137 url 

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

139 ''' 

140 

141try: 

142 MyModel(url='not a url') 

143except ValidationError as e: 

144 print(e) 

145 ''' 

146 1 validation error for MyModel 

147 url 

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

149 ''' 

150``` 

151 

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

153 

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

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

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

157 

158```py 

159from pydantic import BaseModel, HttpUrl 

160 

161class MyModel(BaseModel): 

162 url: HttpUrl 

163 

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

165print(m1.url) 

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

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

168print(m2.url) 

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

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

171print(m3.url) 

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

173``` 

174 

175 

176!!! warning "Underscores in Hostnames" 

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

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

179 

180 To explain this; consider the following two cases: 

181 

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

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

184 

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

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

187 

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

189 (or at least big) company. 

190""" 

191AnyWebsocketUrl = Annotated[Url, UrlConstraints(allowed_schemes=['ws', 'wss'])] 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

192"""A type that will accept any ws or wss URL. 1bcdefgahijklmGHIJKLMnopqrs

193 

194* TLD not required 

195* Host required 

196""" 

197WebsocketUrl = Annotated[Url, UrlConstraints(max_length=2083, allowed_schemes=['ws', 'wss'])] 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

198"""A type that will accept any ws or wss URL. 1bcdefgahijklmGHIJKLMnopqrs

199 

200* TLD not required 

201* Host required 

202* Max length 2083 

203""" 

204FileUrl = Annotated[Url, UrlConstraints(allowed_schemes=['file'])] 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

205"""A type that will accept any file URL. 1bcdefgahijklmGHIJKLMnopqrs

206 

207* Host not required 

208""" 

209FtpUrl = Annotated[Url, UrlConstraints(allowed_schemes=['ftp'])] 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

210"""A type that will accept ftp URL. 1bcdefgahijklmGHIJKLMnopqrs

211 

212* TLD not required 

213* Host required 

214""" 

215PostgresDsn = Annotated[ 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

216 MultiHostUrl, 

217 UrlConstraints( 

218 host_required=True, 

219 allowed_schemes=[ 

220 'postgres', 

221 'postgresql', 

222 'postgresql+asyncpg', 

223 'postgresql+pg8000', 

224 'postgresql+psycopg', 

225 'postgresql+psycopg2', 

226 'postgresql+psycopg2cffi', 

227 'postgresql+py-postgresql', 

228 'postgresql+pygresql', 

229 ], 

230 ), 

231] 

232"""A type that will accept any Postgres DSN. 1bcdefgahijklmGHIJKLMnopqrs

233 

234* User info required 

235* TLD not required 

236* Host required 

237* Supports multiple hosts 

238 

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

240 

241```py 

242from pydantic import ( 

243 BaseModel, 

244 HttpUrl, 

245 PostgresDsn, 

246 ValidationError, 

247 field_validator, 

248) 

249 

250class MyModel(BaseModel): 

251 url: HttpUrl 

252 

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

254 

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

256print(repr(m.url)) 

257#> Url('http://www.example.com/') 

258print(m.url.scheme) 

259#> http 

260print(m.url.host) 

261#> www.example.com 

262print(m.url.port) 

263#> 80 

264 

265class MyDatabaseModel(BaseModel): 

266 db: PostgresDsn 

267 

268 @field_validator('db') 

269 def check_db_name(cls, v): 

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

271 return v 

272 

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

274print(m.db) 

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

276 

277try: 

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

279except ValidationError as e: 

280 print(e) 

281 ''' 

282 1 validation error for MyDatabaseModel 

283 db 

284 Assertion failed, database must be provided 

285 assert (None) 

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

287 ''' 

288``` 

289""" 

290 

291CockroachDsn = Annotated[ 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

292 Url, 

293 UrlConstraints( 

294 host_required=True, 

295 allowed_schemes=[ 

296 'cockroachdb', 

297 'cockroachdb+psycopg2', 

298 'cockroachdb+asyncpg', 

299 ], 

300 ), 

301] 

302"""A type that will accept any Cockroach DSN. 1bcdefgahijklmGHIJKLMnopqrs

303 

304* User info required 

305* TLD not required 

306* Host required 

307""" 

308AmqpDsn = Annotated[Url, UrlConstraints(allowed_schemes=['amqp', 'amqps'])] 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

309"""A type that will accept any AMQP DSN. 1bcdefgahijklmGHIJKLMnopqrs

310 

311* User info required 

312* TLD not required 

313* Host required 

314""" 

315RedisDsn = Annotated[ 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

316 Url, 

317 UrlConstraints(allowed_schemes=['redis', 'rediss'], default_host='localhost', default_port=6379, default_path='/0'), 

318] 

319"""A type that will accept any Redis DSN. 1bcdefgahijklmGHIJKLMnopqrs

320 

321* User info required 

322* TLD not required 

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

324""" 

325MongoDsn = Annotated[MultiHostUrl, UrlConstraints(allowed_schemes=['mongodb', 'mongodb+srv'], default_port=27017)] 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

326"""A type that will accept any MongoDB DSN. 1bcdefgahijklmGHIJKLMnopqrs

327 

328* User info not required 

329* Database name not required 

330* Port not required 

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

332""" 

333KafkaDsn = Annotated[Url, UrlConstraints(allowed_schemes=['kafka'], default_host='localhost', default_port=9092)] 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

334"""A type that will accept any Kafka DSN. 1bcdefgahijklmGHIJKLMnopqrs

335 

336* User info required 

337* TLD not required 

338* Host required 

339""" 

340NatsDsn = Annotated[ 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

341 MultiHostUrl, UrlConstraints(allowed_schemes=['nats', 'tls', 'ws'], default_host='localhost', default_port=4222) 

342] 

343"""A type that will accept any NATS DSN. 1bcdefgahijklmGHIJKLMnopqrs

344 

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

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

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

348More: https://nats.io 

349""" 

350MySQLDsn = Annotated[ 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

351 Url, 

352 UrlConstraints( 

353 allowed_schemes=[ 

354 'mysql', 

355 'mysql+mysqlconnector', 

356 'mysql+aiomysql', 

357 'mysql+asyncmy', 

358 'mysql+mysqldb', 

359 'mysql+pymysql', 

360 'mysql+cymysql', 

361 'mysql+pyodbc', 

362 ], 

363 default_port=3306, 

364 ), 

365] 

366"""A type that will accept any MySQL DSN. 1bcdefgahijklmGHIJKLMnopqrs

367 

368* User info required 

369* TLD not required 

370* Host required 

371""" 

372MariaDBDsn = Annotated[ 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

373 Url, 

374 UrlConstraints( 

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

376 default_port=3306, 

377 ), 

378] 

379"""A type that will accept any MariaDB DSN. 1bcdefgahijklmGHIJKLMnopqrs

380 

381* User info required 

382* TLD not required 

383* Host required 

384""" 

385ClickHouseDsn = Annotated[ 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

386 Url, 

387 UrlConstraints( 

388 allowed_schemes=['clickhouse+native', 'clickhouse+asynch'], 

389 default_host='localhost', 

390 default_port=9000, 

391 ), 

392] 

393"""A type that will accept any ClickHouse DSN. 1bcdefgahijklmGHIJKLMnopqrs

394 

395* User info required 

396* TLD not required 

397* Host required 

398""" 

399 

400 

401def import_email_validator() -> None: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

402 global email_validator 

403 try: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

404 import email_validator 1tuvwbcdefgxayzABhijklmCDEFnopqrs

405 except ImportError as e: 

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

407 if not version('email-validator').partition('.')[0] == '2': 407 ↛ 408line 407 didn't jump to line 408, because the condition on line 407 was never true1tuvwbcdefgxayzABhijklmCDEFnopqrs

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

409 

410 

411if TYPE_CHECKING: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

412 EmailStr = Annotated[str, ...] 

413else: 

414 

415 class EmailStr: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

416 """ 

417 Info: 

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

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

420 

421 ```bash 

422 pip install email-validator 

423 ``` 

424 

425 Validate email addresses. 

426 

427 ```py 

428 from pydantic import BaseModel, EmailStr 

429 

430 class Model(BaseModel): 

431 email: EmailStr 

432 

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

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

435 ``` 

436 """ # noqa: D212 

437 

438 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

439 def __get_pydantic_core_schema__( 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

440 cls, 

441 _source: type[Any], 

442 _handler: GetCoreSchemaHandler, 

443 ) -> core_schema.CoreSchema: 

444 import_email_validator() 1tuvwbcdefgxayzABhijklmCDEFnopqrs

445 return core_schema.no_info_after_validator_function(cls._validate, core_schema.str_schema()) 1tuvwbcdefgxayzABhijklmCDEFnopqrs

446 

447 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

448 def __get_pydantic_json_schema__( 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

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

450 ) -> JsonSchemaValue: 

451 field_schema = handler(core_schema) 1tuvwbcdefgxayzABhijklmCDEFnopqrs

452 field_schema.update(type='string', format='email') 1tuvwbcdefgxayzABhijklmCDEFnopqrs

453 return field_schema 1tuvwbcdefgxayzABhijklmCDEFnopqrs

454 

455 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

456 def _validate(cls, input_value: str, /) -> str: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

457 return validate_email(input_value)[1] 1tuvwbcdefgxayzABhijklmCDEFnopqrs

458 

459 

460class NameEmail(_repr.Representation): 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

461 """ 

462 Info: 

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

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

465 

466 ```bash 

467 pip install email-validator 

468 ``` 

469 

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

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

472 

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

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

475 

476 ```py 

477 from pydantic import BaseModel, NameEmail 

478 

479 class User(BaseModel): 

480 email: NameEmail 

481 

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

483 print(user.email) 

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

485 print(user.email.name) 

486 #> Fred Bloggs 

487 

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

489 print(user.email) 

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

491 print(user.email.name) 

492 #> fred.bloggs 

493 ``` 

494 """ # noqa: D212 

495 

496 __slots__ = 'name', 'email' 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

497 

498 def __init__(self, name: str, email: str): 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

499 self.name = name 1tuvwbcdefgxayzABhijklmCDEFnopqrs

500 self.email = email 1tuvwbcdefgxayzABhijklmCDEFnopqrs

501 

502 def __eq__(self, other: Any) -> bool: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

503 return isinstance(other, NameEmail) and (self.name, self.email) == (other.name, other.email) 1tuvwbcdefgxayzABhijklmCDEFnopqrs

504 

505 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

506 def __get_pydantic_json_schema__( 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

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

508 ) -> JsonSchemaValue: 

509 field_schema = handler(core_schema) 1tuvwbcdefgxayzABhijklmCDEFnopqrs

510 field_schema.update(type='string', format='name-email') 1tuvwbcdefgxayzABhijklmCDEFnopqrs

511 return field_schema 1tuvwbcdefgxayzABhijklmCDEFnopqrs

512 

513 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

514 def __get_pydantic_core_schema__( 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

515 cls, 

516 _source: type[Any], 

517 _handler: GetCoreSchemaHandler, 

518 ) -> core_schema.CoreSchema: 

519 import_email_validator() 1tuvwbcdefgxayzABhijklmCDEFnopqrs

520 

521 return core_schema.no_info_after_validator_function( 1tuvwbcdefgxayzABhijklmCDEFnopqrs

522 cls._validate, 

523 core_schema.json_or_python_schema( 

524 json_schema=core_schema.str_schema(), 

525 python_schema=core_schema.union_schema( 

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

527 custom_error_type='name_email_type', 

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

529 ), 

530 serialization=core_schema.to_string_ser_schema(), 

531 ), 

532 ) 

533 

534 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

535 def _validate(cls, input_value: Self | str, /) -> Self: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

536 if isinstance(input_value, str): 1tuvwbcdefgxayzABhijklmCDEFnopqrs

537 name, email = validate_email(input_value) 1tuvwbcdefgxayzABhijklmCDEFnopqrs

538 return cls(name, email) 1tuvwbcdefgxayzABhijklmCDEFnopqrs

539 else: 

540 return input_value 1tuvwbcdefgxayzABhijklmCDEFnopqrs

541 

542 def __str__(self) -> str: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

543 if '@' in self.name: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

544 return f'"{self.name}" <{self.email}>' 1tuvwbcdefgxayzABhijklmCDEFnopqrs

545 

546 return f'{self.name} <{self.email}>' 1tuvwbcdefgxayzABhijklmCDEFnopqrs

547 

548 

549class IPvAnyAddress: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

550 """Validate an IPv4 or IPv6 address. 

551 

552 ```py 

553 from pydantic import BaseModel 

554 from pydantic.networks import IPvAnyAddress 

555 

556 class IpModel(BaseModel): 

557 ip: IPvAnyAddress 

558 

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

560 #> ip=IPv4Address('127.0.0.1') 

561 

562 try: 

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

564 except ValueError as e: 

565 print(e.errors()) 

566 ''' 

567 [ 

568 { 

569 'type': 'ip_any_address', 

570 'loc': ('ip',), 

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

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

573 } 

574 ] 

575 ''' 

576 ``` 

577 """ 

578 

579 __slots__ = () 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

580 

581 def __new__(cls, value: Any) -> IPv4Address | IPv6Address: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

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

583 try: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

584 return IPv4Address(value) 1tuvwbcdefgxayzABhijklmCDEFnopqrs

585 except ValueError: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

586 pass 1tuvwbcdefgxayzABhijklmCDEFnopqrs

587 

588 try: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

589 return IPv6Address(value) 1tuvwbcdefgxayzABhijklmCDEFnopqrs

590 except ValueError: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

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

592 

593 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

594 def __get_pydantic_json_schema__( 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

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

596 ) -> JsonSchemaValue: 

597 field_schema = {} 1tuvwbcdefgxayzABhijklmCDEFnopqrs

598 field_schema.update(type='string', format='ipvanyaddress') 1tuvwbcdefgxayzABhijklmCDEFnopqrs

599 return field_schema 1tuvwbcdefgxayzABhijklmCDEFnopqrs

600 

601 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

602 def __get_pydantic_core_schema__( 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

603 cls, 

604 _source: type[Any], 

605 _handler: GetCoreSchemaHandler, 

606 ) -> core_schema.CoreSchema: 

607 return core_schema.no_info_plain_validator_function( 1tuvwbcdefgxayzABhijklmCDEFnopqrs

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

609 ) 

610 

611 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

612 def _validate(cls, input_value: Any, /) -> IPv4Address | IPv6Address: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

613 return cls(input_value) # type: ignore[return-value] 1tuvwbcdefgxayzABhijklmCDEFnopqrs

614 

615 

616class IPvAnyInterface: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

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

618 

619 __slots__ = () 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

620 

621 def __new__(cls, value: NetworkType) -> IPv4Interface | IPv6Interface: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

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

623 try: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

624 return IPv4Interface(value) 1tuvwbcdefgxayzABhijklmCDEFnopqrs

625 except ValueError: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

626 pass 1tuvwbcdefgxayzABhijklmCDEFnopqrs

627 

628 try: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

629 return IPv6Interface(value) 1tuvwbcdefgxayzABhijklmCDEFnopqrs

630 except ValueError: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

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

632 

633 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

634 def __get_pydantic_json_schema__( 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

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

636 ) -> JsonSchemaValue: 

637 field_schema = {} 1tuvwbcdefgxayzABhijklmCDEFnopqrs

638 field_schema.update(type='string', format='ipvanyinterface') 1tuvwbcdefgxayzABhijklmCDEFnopqrs

639 return field_schema 1tuvwbcdefgxayzABhijklmCDEFnopqrs

640 

641 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

642 def __get_pydantic_core_schema__( 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

643 cls, 

644 _source: type[Any], 

645 _handler: GetCoreSchemaHandler, 

646 ) -> core_schema.CoreSchema: 

647 return core_schema.no_info_plain_validator_function( 1tuvwbcdefgxayzABhijklmCDEFnopqrs

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

649 ) 

650 

651 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

652 def _validate(cls, input_value: NetworkType, /) -> IPv4Interface | IPv6Interface: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

653 return cls(input_value) # type: ignore[return-value] 1tuvwbcdefgxayzABhijklmCDEFnopqrs

654 

655 

656IPvAnyNetworkType: TypeAlias = 'IPv4Network | IPv6Network' 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

657 

658if TYPE_CHECKING: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

659 IPvAnyNetwork = IPvAnyNetworkType 

660else: 

661 

662 class IPvAnyNetwork: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

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

664 

665 __slots__ = () 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

666 

667 def __new__(cls, value: NetworkType) -> IPvAnyNetworkType: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

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

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

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

671 try: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

672 return IPv4Network(value) 1tuvwbcdefgxayzABhijklmCDEFnopqrs

673 except ValueError: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

674 pass 1tuvwbcdefgxayzABhijklmCDEFnopqrs

675 

676 try: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

677 return IPv6Network(value) 1tuvwbcdefgxayzABhijklmCDEFnopqrs

678 except ValueError: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

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

680 

681 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

682 def __get_pydantic_json_schema__( 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

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

684 ) -> JsonSchemaValue: 

685 field_schema = {} 1tuvwbcdefgxayzABhijklmCDEFnopqrs

686 field_schema.update(type='string', format='ipvanynetwork') 1tuvwbcdefgxayzABhijklmCDEFnopqrs

687 return field_schema 1tuvwbcdefgxayzABhijklmCDEFnopqrs

688 

689 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

690 def __get_pydantic_core_schema__( 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

691 cls, 

692 _source: type[Any], 

693 _handler: GetCoreSchemaHandler, 

694 ) -> core_schema.CoreSchema: 

695 return core_schema.no_info_plain_validator_function( 1tuvwbcdefgxayzABhijklmCDEFnopqrs

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

697 ) 

698 

699 @classmethod 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

700 def _validate(cls, input_value: NetworkType, /) -> IPvAnyNetworkType: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

701 return cls(input_value) # type: ignore[return-value] 1tuvwbcdefgxayzABhijklmCDEFnopqrs

702 

703 

704def _build_pretty_email_regex() -> re.Pattern[str]: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

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

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

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

708 email_group = r'<\s*(.+)\s*>' 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

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

710 

711 

712pretty_email_regex = _build_pretty_email_regex() 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

713 

714MAX_EMAIL_LENGTH = 2048 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

715"""Maximum length for an email. 1bcdefgahijklmGHIJKLMnopqrs

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

717""" 

718 

719 

720def validate_email(value: str) -> tuple[str, str]: 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs

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

722 

723 Note: 

724 Note that: 

725 

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

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

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

729 """ 

730 if email_validator is None: 730 ↛ 731line 730 didn't jump to line 731, because the condition on line 730 was never true1tuvwbcdefgxayzABhijklmCDEFnopqrs

731 import_email_validator() 

732 

733 if len(value) > MAX_EMAIL_LENGTH: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

734 raise PydanticCustomError( 1tuvwbcdefgxayzABhijklmCDEFnopqrs

735 'value_error', 

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

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

738 ) 

739 

740 m = pretty_email_regex.fullmatch(value) 1tuvwbcdefgxayzABhijklmCDEFnopqrs

741 name: str | None = None 1tuvwbcdefgxayzABhijklmCDEFnopqrs

742 if m: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

743 unquoted_name, quoted_name, value = m.groups() 1tuvwbcdefgxayzABhijklmCDEFnopqrs

744 name = unquoted_name or quoted_name 1tuvwbcdefgxayzABhijklmCDEFnopqrs

745 

746 email = value.strip() 1tuvwbcdefgxayzABhijklmCDEFnopqrs

747 

748 try: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

749 parts = email_validator.validate_email(email, check_deliverability=False) 1tuvwbcdefgxayzABhijklmCDEFnopqrs

750 except email_validator.EmailNotValidError as e: 1tuvwbcdefgxayzABhijklmCDEFnopqrs

751 raise PydanticCustomError( 1tuvwbcdefgxayzABhijklmCDEFnopqrs

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

753 ) from e 

754 

755 email = parts.normalized 1tuvwbcdefgxayzABhijklmCDEFnopqrs

756 assert email is not None 1tuvwbcdefgxayzABhijklmCDEFnopqrs

757 name = name or parts.local_part 1tuvwbcdefgxayzABhijklmCDEFnopqrs

758 return name, email 1tuvwbcdefgxayzABhijklmCDEFnopqrs

759 

760 

761__getattr__ = getattr_migration(__name__) 1tuvwbcdefgxayzABhijklmNOGHIJKLMCDEFnopqrs