Coverage for /pythoncovmergedfiles/medio/medio/src/pydantic/pydantic/networks.py: 60%

1from __future__ import annotations as _annotations 

2 

3import dataclasses as _dataclasses 

4import re 

5from ipaddress import IPv4Address, IPv4Interface, IPv4Network, IPv6Address, IPv6Interface, IPv6Network 

6from typing import TYPE_CHECKING, Any, Callable 

7 

8from pydantic_core import MultiHostUrl, PydanticCustomError, Url, core_schema 

9from typing_extensions import Annotated, TypeAlias 

10 

11from ._internal import _fields, _repr 

12from ._internal._json_schema_shared import GetJsonSchemaHandler 

13from ._migration import getattr_migration 

14from .json_schema import JsonSchemaValue 

15 

16if TYPE_CHECKING: 

17 import email_validator 

18 

19 NetworkType: TypeAlias = 'str | bytes | int | tuple[str | bytes | int, str | int]' 

20 

21else: 

22 email_validator = None 

23 

24 

25__all__ = [ 

26 'AnyUrl', 

27 'AnyHttpUrl', 

28 'FileUrl', 

29 'HttpUrl', 

30 'UrlConstraints', 

31 'EmailStr', 

32 'NameEmail', 

33 'IPvAnyAddress', 

34 'IPvAnyInterface', 

35 'IPvAnyNetwork', 

36 'PostgresDsn', 

37 'CockroachDsn', 

38 'AmqpDsn', 

39 'RedisDsn', 

40 'MongoDsn', 

41 'KafkaDsn', 

42 'validate_email', 

43 'MySQLDsn', 

44 'MariaDBDsn', 

45] 

46 

47 

48@_dataclasses.dataclass 

49class UrlConstraints(_fields.PydanticMetadata): 

50 max_length: int | None = None 

51 allowed_schemes: list[str] | None = None 

52 host_required: bool | None = None 

53 default_host: str | None = None 

54 default_port: int | None = None 

55 default_path: str | None = None 

56 

57 

58AnyUrl = Url 

59# host_required is false because all schemes are "special" so host is required by rust-url automatically 

60AnyHttpUrl = Annotated[Url, UrlConstraints(allowed_schemes=['http', 'https'])] 

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

62FileUrl = Annotated[Url, UrlConstraints(allowed_schemes=['file'])] 

63PostgresDsn = Annotated[ 

64 MultiHostUrl, 

65 UrlConstraints( 

66 host_required=True, 

67 allowed_schemes=[ 

68 'postgres', 

69 'postgresql', 

70 'postgresql+asyncpg', 

71 'postgresql+pg8000', 

72 'postgresql+psycopg', 

73 'postgresql+psycopg2', 

74 'postgresql+psycopg2cffi', 

75 'postgresql+py-postgresql', 

76 'postgresql+pygresql', 

77 ], 

78 ), 

79] 

80 

81CockroachDsn = Annotated[ 

82 Url, 

83 UrlConstraints( 

84 host_required=True, 

85 allowed_schemes=[ 

86 'cockroachdb', 

87 'cockroachdb+psycopg2', 

88 'cockroachdb+asyncpg', 

89 ], 

90 ), 

91] 

92AmqpDsn = Annotated[Url, UrlConstraints(allowed_schemes=['amqp', 'amqps'])] 

93RedisDsn = Annotated[ 

94 Url, 

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

96] 

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

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

99MySQLDsn = Annotated[ 

100 Url, 

101 UrlConstraints( 

102 allowed_schemes=[ 

103 'mysql', 

104 'mysql+mysqlconnector', 

105 'mysql+aiomysql', 

106 'mysql+asyncmy', 

107 'mysql+mysqldb', 

108 'mysql+pymysql', 

109 'mysql+cymysql', 

110 'mysql+pyodbc', 

111 ], 

112 default_port=3306, 

113 ), 

114] 

115MariaDBDsn = Annotated[ 

116 Url, 

117 UrlConstraints( 

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

119 default_port=3306, 

120 ), 

121] 

122 

123 

124def import_email_validator() -> None: 

125 global email_validator 

126 try: 

127 import email_validator 

128 except ImportError as e: 

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

130 

131 

132if TYPE_CHECKING: 

133 EmailStr = Annotated[str, ...] 

134else: 

135 

136 class EmailStr: 

137 @classmethod 

138 def __get_pydantic_core_schema__( 

139 cls, source: type[Any], handler: Callable[[Any], core_schema.CoreSchema] 

140 ) -> core_schema.CoreSchema: 

141 import_email_validator() 

142 return core_schema.general_after_validator_function(cls.validate, core_schema.str_schema()) 

143 

144 @classmethod 

145 def __get_pydantic_json_schema__( 

146 cls, core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler 

147 ) -> JsonSchemaValue: 

148 field_schema = handler(core_schema) 

149 field_schema.update(type='string', format='email') 

150 return field_schema 

151 

152 @classmethod 

153 def validate(cls, __input_value: str, _: core_schema.ValidationInfo) -> str: 

154 return validate_email(__input_value)[1] 

155 

156 

157class NameEmail(_repr.Representation): 

158 __slots__ = 'name', 'email' 

159 

160 def __init__(self, name: str, email: str): 

161 self.name = name 

162 self.email = email 

163 

164 def __eq__(self, other: Any) -> bool: 

165 return isinstance(other, NameEmail) and (self.name, self.email) == (other.name, other.email) 

166 

167 @classmethod 

168 def __get_pydantic_json_schema__( 

169 cls, core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler 

170 ) -> JsonSchemaValue: 

171 field_schema = handler(core_schema) 

172 field_schema.update(type='string', format='name-email') 

173 return field_schema 

174 

175 @classmethod 

176 def __get_pydantic_core_schema__( 

177 cls, source: type[Any], handler: Callable[[Any], core_schema.CoreSchema] 

178 ) -> core_schema.CoreSchema: 

179 import_email_validator() 

180 return core_schema.general_after_validator_function( 

181 cls._validate, 

182 core_schema.union_schema([core_schema.is_instance_schema(cls), core_schema.str_schema()]), 

183 serialization=core_schema.to_string_ser_schema(), 

184 ) 

185 

186 @classmethod 

187 def _validate(cls, __input_value: NameEmail | str, _: core_schema.ValidationInfo) -> NameEmail: 

188 if isinstance(__input_value, cls): 

189 return __input_value 

190 else: 

191 name, email = validate_email(__input_value) # type: ignore[arg-type] 

192 return cls(name, email) 

193 

194 def __str__(self) -> str: 

195 return f'{self.name} <{self.email}>' 

196 

197 

198class IPvAnyAddress: 

199 __slots__ = () 

200 

201 def __new__(cls, value: Any) -> IPv4Address | IPv6Address: # type: ignore[misc] 

202 try: 

203 return IPv4Address(value) 

204 except ValueError: 

205 pass 

206 

207 try: 

208 return IPv6Address(value) 

209 except ValueError: 

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

211 

212 @classmethod 

213 def __get_pydantic_json_schema__( 

214 cls, core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler 

215 ) -> JsonSchemaValue: 

216 field_schema = {} 

217 field_schema.update(type='string', format='ipvanyaddress') 

218 return field_schema 

219 

220 @classmethod 

221 def __get_pydantic_core_schema__( 

222 cls, source: type[Any], handler: Callable[[Any], core_schema.CoreSchema] 

223 ) -> core_schema.CoreSchema: 

224 return core_schema.general_plain_validator_function(cls._validate) 

225 

226 @classmethod 

227 def _validate(cls, __input_value: Any, _: core_schema.ValidationInfo) -> IPv4Address | IPv6Address: 

228 return cls(__input_value) # type: ignore[return-value] 

229 

230 

231class IPvAnyInterface: 

232 __slots__ = () 

233 

234 def __new__(cls, value: NetworkType) -> IPv4Interface | IPv6Interface: # type: ignore[misc] 

235 try: 

236 return IPv4Interface(value) 

237 except ValueError: 

238 pass 

239 

240 try: 

241 return IPv6Interface(value) 

242 except ValueError: 

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

244 

245 @classmethod 

246 def __get_pydantic_json_schema__( 

247 cls, core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler 

248 ) -> JsonSchemaValue: 

249 field_schema = {} 

250 field_schema.update(type='string', format='ipvanyinterface') 

251 return field_schema 

252 

253 @classmethod 

254 def __get_pydantic_core_schema__( 

255 cls, source: type[Any], handler: Callable[[Any], core_schema.CoreSchema] 

256 ) -> core_schema.CoreSchema: 

257 return core_schema.general_plain_validator_function(cls._validate) 

258 

259 @classmethod 

260 def _validate(cls, __input_value: NetworkType, _: core_schema.ValidationInfo) -> IPv4Interface | IPv6Interface: 

261 return cls(__input_value) # type: ignore[return-value] 

262 

263 

264class IPvAnyNetwork: 

265 __slots__ = () 

266 

267 def __new__(cls, value: NetworkType) -> IPv4Network | IPv6Network: # type: ignore[misc] 

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

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

270 try: 

271 return IPv4Network(value) 

272 except ValueError: 

273 pass 

274 

275 try: 

276 return IPv6Network(value) 

277 except ValueError: 

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

279 

280 @classmethod 

281 def __get_pydantic_json_schema__( 

282 cls, core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler 

283 ) -> JsonSchemaValue: 

284 field_schema = {} 

285 field_schema.update(type='string', format='ipvanynetwork') 

286 return field_schema 

287 

288 @classmethod 

289 def __get_pydantic_core_schema__( 

290 cls, source: type[Any], handler: Callable[[Any], core_schema.CoreSchema] 

291 ) -> core_schema.CoreSchema: 

292 return core_schema.general_plain_validator_function(cls._validate) 

293 

294 @classmethod 

295 def _validate(cls, __input_value: NetworkType, _: core_schema.ValidationInfo) -> IPv4Network | IPv6Network: 

296 return cls(__input_value) # type: ignore[return-value] 

297 

298 

299pretty_email_regex = re.compile(r' *([\w ]*?) *<(.+?)> *') 

300 

301 

302def validate_email(value: str) -> tuple[str, str]: 

303 """ 

304 Email address validation using https://pypi.org/project/email-validator/ 

305 

306 Notes: 

307 * raw ip address (literal) domain parts are not allowed. 

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

309 * spaces are striped from the beginning and end of addresses but no error is raised 

310 """ 

311 if email_validator is None: 

312 import_email_validator() 

313 

314 m = pretty_email_regex.fullmatch(value) 

315 name: str | None = None 

316 if m: 

317 name, value = m.groups() 

318 

319 email = value.strip() 

320 

321 try: 

322 parts = email_validator.validate_email(email, check_deliverability=False) 

323 except email_validator.EmailNotValidError as e: 

324 raise PydanticCustomError( 

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

326 ) from e 

327 

328 return name or parts['local'], parts['email'] 

329 

330 

331__getattr__ = getattr_migration(__name__)