Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Validator functions for standard library types.
3Import of this module is deferred since it contains imports of many standard library modules.
4"""
6from __future__ import annotations as _annotations
8import collections.abc
9import math
10import re
11import sys
12import typing
13from collections.abc import Callable, Sequence
14from decimal import Decimal
15from ipaddress import IPv4Address, IPv4Interface, IPv4Network, IPv6Address, IPv6Interface, IPv6Network
16from typing import Any, TypeAlias, TypeVar, cast
17from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
19from pydantic_core import PydanticCustomError, PydanticKnownError, core_schema
20from typing_extensions import get_args, get_origin # noqa: UP035
21from typing_inspection import typing_objects
23from pydantic._internal._import_utils import import_cached_field_info
24from pydantic._internal._typing_extra import is_namedtuple
25from pydantic.errors import PydanticSchemaGenerationError
28def sequence_validator(
29 input_value: Sequence[Any],
30 /,
31 validator: core_schema.ValidatorFunctionWrapHandler,
32) -> Sequence[Any]:
33 """Validator for `Sequence` types, isinstance(v, Sequence) has already been called."""
34 value_type = type(input_value)
36 # We don't accept any plain string as a sequence
37 # Relevant issue: https://github.com/pydantic/pydantic/issues/5595
38 if issubclass(value_type, (str, bytes)):
39 raise PydanticCustomError(
40 'sequence_str',
41 "'{type_name}' instances are not allowed as a Sequence value",
42 {'type_name': value_type.__name__},
43 )
45 # TODO: refactor sequence validation to validate with either a list or a tuple
46 # schema, depending on the type of the value.
47 # Additionally, we should be able to remove one of either this validator or the
48 # SequenceValidator in _std_types_schema.py (preferably this one, while porting over some logic).
49 # Effectively, a refactor for sequence validation is needed.
50 if value_type is tuple:
51 input_value = list(input_value)
53 v_list = validator(input_value)
55 # the rest of the logic is just re-creating the original type from `v_list`
56 if value_type is list:
57 return v_list
58 elif issubclass(value_type, range):
59 # return the list as we probably can't re-create the range
60 return v_list
61 elif value_type is tuple:
62 return tuple(v_list)
63 elif is_namedtuple(value_type):
64 return value_type(*v_list)
65 else:
66 # best guess at how to re-create the original type, more custom construction logic might be required
67 return value_type(v_list) # type: ignore[call-arg]
70def import_string(value: Any) -> Any:
71 if isinstance(value, str):
72 try:
73 return _import_string_logic(value)
74 except ImportError as e:
75 raise PydanticCustomError('import_error', 'Invalid python path: {error}', {'error': str(e)}) from e
76 else:
77 # otherwise we just return the value and let the next validator do the rest of the work
78 return value
81def _import_string_logic(dotted_path: str) -> Any:
82 """Inspired by uvicorn — dotted paths should include a colon before the final item if that item is not a module.
83 (This is necessary to distinguish between a submodule and an attribute when there is a conflict.)
85 If the dotted path does not include a colon and the final item is not a valid module, importing as an attribute
86 rather than a submodule will be attempted automatically.
88 So, for example, the following values of `dotted_path` result in the following returned values:
89 * 'collections': <module 'collections'>
90 * 'collections.abc': <module 'collections.abc'>
91 * 'collections.abc:Mapping': <class 'collections.abc.Mapping'>
92 * `collections.abc.Mapping`: <class 'collections.abc.Mapping'> (though this is a bit slower than the previous line)
94 An error will be raised under any of the following scenarios:
95 * `dotted_path` contains more than one colon (e.g., 'collections:abc:Mapping')
96 * the substring of `dotted_path` before the colon is not a valid module in the environment (e.g., '123:Mapping')
97 * the substring of `dotted_path` after the colon is not an attribute of the module (e.g., 'collections:abc123')
98 """
99 from importlib import import_module
101 components = dotted_path.strip().split(':')
102 if len(components) > 2:
103 raise ImportError(f"Import strings should have at most one ':'; received {dotted_path!r}")
104 attribute = None
105 if len(components) == 2:
106 attribute = components[1]
107 module_path = components[0]
108 if not module_path:
109 raise ImportError(f'Import strings should have a nonempty module name; received {dotted_path!r}')
111 try:
112 module = import_module(module_path)
113 except ModuleNotFoundError:
114 if attribute is None and '.' in module_path:
115 # Try interpreting the final dotted segment as an attribute, not a submodule
116 maybe_module_path, maybe_attribute = module_path.rsplit('.', 1)
118 try:
119 return _import_string_logic(f'{maybe_module_path}:{maybe_attribute}')
120 except ImportError:
121 pass
122 raise
124 if attribute is not None:
125 try:
126 return getattr(module, attribute)
127 except AttributeError as e:
128 raise ImportError(f'cannot import name {attribute!r} from {module_path!r}') from e
129 else:
130 return module
133def pattern_either_validator(input_value: Any, /) -> re.Pattern[Any]:
134 if isinstance(input_value, re.Pattern):
135 return input_value
136 elif isinstance(input_value, (str, bytes)):
137 # todo strict mode
138 return compile_pattern(input_value) # type: ignore
139 else:
140 raise PydanticCustomError('pattern_type', 'Input should be a valid pattern')
143def pattern_str_validator(input_value: Any, /) -> re.Pattern[str]:
144 if isinstance(input_value, re.Pattern):
145 if isinstance(input_value.pattern, str):
146 return input_value
147 else:
148 raise PydanticCustomError('pattern_str_type', 'Input should be a string pattern')
149 elif isinstance(input_value, str):
150 return compile_pattern(input_value)
151 elif isinstance(input_value, bytes):
152 raise PydanticCustomError('pattern_str_type', 'Input should be a string pattern')
153 else:
154 raise PydanticCustomError('pattern_type', 'Input should be a valid pattern')
157def pattern_bytes_validator(input_value: Any, /) -> re.Pattern[bytes]:
158 if isinstance(input_value, re.Pattern):
159 if isinstance(input_value.pattern, bytes):
160 return input_value
161 else:
162 raise PydanticCustomError('pattern_bytes_type', 'Input should be a bytes pattern')
163 elif isinstance(input_value, bytes):
164 return compile_pattern(input_value)
165 elif isinstance(input_value, str):
166 raise PydanticCustomError('pattern_bytes_type', 'Input should be a bytes pattern')
167 else:
168 raise PydanticCustomError('pattern_type', 'Input should be a valid pattern')
171PatternType = TypeVar('PatternType', str, bytes)
174def compile_pattern(pattern: PatternType) -> re.Pattern[PatternType]:
175 try:
176 return re.compile(pattern)
177 except re.error:
178 raise PydanticCustomError('pattern_regex', 'Input should be a valid regular expression')
181def ip_v4_address_validator(input_value: Any, /) -> IPv4Address:
182 if isinstance(input_value, IPv4Address):
183 return input_value
185 try:
186 return IPv4Address(input_value)
187 except ValueError:
188 raise PydanticCustomError('ip_v4_address', 'Input is not a valid IPv4 address')
191def ip_v6_address_validator(input_value: Any, /) -> IPv6Address:
192 if isinstance(input_value, IPv6Address):
193 return input_value
195 try:
196 return IPv6Address(input_value)
197 except ValueError:
198 raise PydanticCustomError('ip_v6_address', 'Input is not a valid IPv6 address')
201def ip_v4_network_validator(input_value: Any, /) -> IPv4Network:
202 """Assume IPv4Network initialised with a default `strict` argument.
204 See more:
205 https://docs.python.org/library/ipaddress.html#ipaddress.IPv4Network
206 """
207 if isinstance(input_value, IPv4Network):
208 return input_value
210 try:
211 return IPv4Network(input_value)
212 except ValueError:
213 raise PydanticCustomError('ip_v4_network', 'Input is not a valid IPv4 network')
216def ip_v6_network_validator(input_value: Any, /) -> IPv6Network:
217 """Assume IPv6Network initialised with a default `strict` argument.
219 See more:
220 https://docs.python.org/library/ipaddress.html#ipaddress.IPv6Network
221 """
222 if isinstance(input_value, IPv6Network):
223 return input_value
225 try:
226 return IPv6Network(input_value)
227 except ValueError:
228 raise PydanticCustomError('ip_v6_network', 'Input is not a valid IPv6 network')
231def ip_v4_interface_validator(input_value: Any, /) -> IPv4Interface:
232 if isinstance(input_value, IPv4Interface):
233 return input_value
235 try:
236 return IPv4Interface(input_value)
237 except ValueError:
238 raise PydanticCustomError('ip_v4_interface', 'Input is not a valid IPv4 interface')
241def ip_v6_interface_validator(input_value: Any, /) -> IPv6Interface:
242 if isinstance(input_value, IPv6Interface):
243 return input_value
245 try:
246 return IPv6Interface(input_value)
247 except ValueError:
248 raise PydanticCustomError('ip_v6_interface', 'Input is not a valid IPv6 interface')
251def forbid_inf_nan_check(x: Any) -> Any:
252 if not math.isfinite(x):
253 raise PydanticKnownError('finite_number')
254 return x
257def _safe_repr(v: Any) -> int | float | str:
258 """The context argument for `PydanticKnownError` requires a number or str type, so we do a simple repr() coercion for types like timedelta.
260 See tests/test_types.py::test_annotated_metadata_any_order for some context.
261 """
262 if isinstance(v, (int, float, str)):
263 return v
264 return repr(v)
267def greater_than_validator(x: Any, gt: Any) -> Any:
268 try:
269 if not (x > gt):
270 raise PydanticKnownError('greater_than', {'gt': _safe_repr(gt)})
271 return x
272 except TypeError:
273 raise TypeError(f"Unable to apply constraint 'gt' to supplied value {x}")
276def greater_than_or_equal_validator(x: Any, ge: Any) -> Any:
277 try:
278 if not (x >= ge):
279 raise PydanticKnownError('greater_than_equal', {'ge': _safe_repr(ge)})
280 return x
281 except TypeError:
282 raise TypeError(f"Unable to apply constraint 'ge' to supplied value {x}")
285def less_than_validator(x: Any, lt: Any) -> Any:
286 try:
287 if not (x < lt):
288 raise PydanticKnownError('less_than', {'lt': _safe_repr(lt)})
289 return x
290 except TypeError:
291 raise TypeError(f"Unable to apply constraint 'lt' to supplied value {x}")
294def less_than_or_equal_validator(x: Any, le: Any) -> Any:
295 try:
296 if not (x <= le):
297 raise PydanticKnownError('less_than_equal', {'le': _safe_repr(le)})
298 return x
299 except TypeError:
300 raise TypeError(f"Unable to apply constraint 'le' to supplied value {x}")
303def multiple_of_validator(x: Any, multiple_of: Any) -> Any:
304 try:
305 if x % multiple_of:
306 raise PydanticKnownError('multiple_of', {'multiple_of': _safe_repr(multiple_of)})
307 return x
308 except TypeError:
309 raise TypeError(f"Unable to apply constraint 'multiple_of' to supplied value {x}")
312def min_length_validator(x: Any, min_length: Any) -> Any:
313 try:
314 if not (len(x) >= min_length):
315 raise PydanticKnownError(
316 'too_short', {'field_type': 'Value', 'min_length': min_length, 'actual_length': len(x)}
317 )
318 return x
319 except TypeError:
320 raise TypeError(f"Unable to apply constraint 'min_length' to supplied value {x}")
323def max_length_validator(x: Any, max_length: Any) -> Any:
324 try:
325 if len(x) > max_length:
326 raise PydanticKnownError(
327 'too_long',
328 {'field_type': 'Value', 'max_length': max_length, 'actual_length': len(x)},
329 )
330 return x
331 except TypeError:
332 raise TypeError(f"Unable to apply constraint 'max_length' to supplied value {x}")
335def _count_digits(num_digits: int, exponent: int) -> tuple[int, int]:
336 if exponent >= 0:
337 # A positive exponent adds that many trailing zeros
338 # Ex: digit_tuple=(1, 2, 3), exponent=2 -> 12300 -> 0 decimal places, 5 digits
339 return 0, num_digits + exponent
340 else:
341 # If the absolute value of the negative exponent is larger than the
342 # number of digits, then it's the same as the number of digits,
343 # because it'll consume all the digits in digit_tuple and then
344 # add abs(exponent) - len(digit_tuple) leading zeros after the decimal point.
345 # Ex: digit_tuple=(1, 2, 3), exponent=-2 -> 1.23 -> 2 decimal places, 3 digits
346 # Ex: digit_tuple=(1, 2, 3), exponent=-4 -> 0.0123 -> 4 decimal places, 4 digits
347 decimal_places = abs(exponent)
348 return decimal_places, max(num_digits, decimal_places)
351def _extract_decimal_digits_info(decimal: Decimal) -> tuple[tuple[int, int], tuple[int, int]]:
352 """Compute the number of decimal places and total digits for a given [`Decimal`][decimal.Decimal] instance.
354 Returns a tuple of two `(decimal_places, num_digits)` pairs: the first one for the decimal as is,
355 the second one for its normalized form (i.e. with trailing zeros stripped).
356 Example: `Decimal('1.230')` returns `((3, 4), (2, 3))`.
358 Trailing zeros are stripped by hand rather than by calling [`Decimal.normalize()`][decimal.Decimal.normalize],
359 as the latter also rounds the value to the precision of the current decimal context.
360 """
361 try:
362 decimal_tuple = decimal.as_tuple()
364 assert isinstance(decimal_tuple.exponent, int)
366 exponent = decimal_tuple.exponent
367 digits = decimal_tuple.digits
368 num_digits = len(digits)
370 trailing_zeros = 0
371 for digit in reversed(digits):
372 if digit != 0:
373 break
374 trailing_zeros += 1
376 if trailing_zeros == num_digits:
377 # All digits are zero: `Decimal.normalize()` canonicalizes any zero to `0E0`.
378 normalized = (0, 1)
379 else:
380 # Each stripped trailing zero is compensated by incrementing the exponent.
381 normalized = _count_digits(num_digits - trailing_zeros, exponent + trailing_zeros)
383 return _count_digits(num_digits, exponent), normalized
384 except (AssertionError, AttributeError):
385 raise TypeError(f'Unable to extract decimal digits info from supplied value {decimal}')
388def max_digits_validator(x: Any, max_digits: Any) -> Any:
389 try:
390 (_, num_digits), (_, normalized_num_digits) = _extract_decimal_digits_info(x)
391 if (num_digits > max_digits) and (normalized_num_digits > max_digits):
392 raise PydanticKnownError(
393 'decimal_max_digits',
394 {'max_digits': max_digits},
395 )
396 return x
397 except TypeError:
398 raise TypeError(f"Unable to apply constraint 'max_digits' to supplied value {x}")
401def decimal_places_validator(x: Any, decimal_places: Any, max_digits: Any = None) -> Any:
402 try:
403 (decimal_places_, num_digits), (normalized_decimal_places, normalized_num_digits) = (
404 _extract_decimal_digits_info(x)
405 )
406 if (decimal_places_ > decimal_places) and (normalized_decimal_places > decimal_places):
407 raise PydanticKnownError(
408 'decimal_max_places',
409 {'decimal_places': decimal_places},
410 )
411 if max_digits is not None:
412 max_whole_digits = max(max_digits - decimal_places, 0)
413 if (num_digits - decimal_places_ > max_whole_digits) and (
414 normalized_num_digits - normalized_decimal_places > max_whole_digits
415 ):
416 raise PydanticKnownError(
417 'decimal_whole_digits',
418 {'whole_digits': max_whole_digits},
419 )
420 return x
421 except TypeError:
422 raise TypeError(f"Unable to apply constraint 'decimal_places' to supplied value {x}")
425def defaultdict_validator(
426 input_value: Any, handler: core_schema.ValidatorFunctionWrapHandler, default_default_factory: Callable[[], Any]
427) -> collections.defaultdict[Any, Any]:
428 if isinstance(input_value, collections.defaultdict):
429 default_factory = input_value.default_factory
430 return collections.defaultdict(default_factory, handler(input_value))
431 else:
432 return collections.defaultdict(default_default_factory, handler(input_value))
435_defaultdict_allowed_default_types: dict[type[Any], type[Any]] = {
436 float: float,
437 int: int,
438 str: str,
439 bool: bool,
440 tuple: tuple,
441 collections.abc.Sequence: tuple,
442 collections.abc.MutableSequence: list,
443 list: list,
444 set: set,
445 collections.abc.MutableSet: set,
446 frozenset: frozenset,
447 collections.abc.Set: frozenset,
448 dict: dict,
449 collections.abc.Mapping: dict,
450 collections.abc.MutableMapping: dict,
451 collections.OrderedDict: collections.OrderedDict,
452 collections.Counter: collections.Counter,
453}
455if sys.version_info >= (3, 15):
456 _defaultdict_allowed_default_types[collections.abc.Mapping] = frozendict # noqa: F821
457 _defaultdict_allowed_default_types[frozendict] = frozendict # noqa: F821
460def get_defaultdict_default_default_factory(values_source_type: Any) -> Callable[[], Any]:
461 FieldInfo = import_cached_field_info()
463 values_type_origin = get_origin(values_source_type)
465 def infer_default() -> Callable[[], Any]:
466 values_type = values_type_origin or values_source_type
467 instructions = 'set using `DefaultDict[..., Annotated[..., Field(default_factory=...)]]`'
468 if typing_objects.is_typevar(values_type):
470 def type_var_default_factory() -> None:
471 raise RuntimeError(
472 'Generic defaultdict cannot be used without a concrete value type or an'
473 ' explicit default factory, ' + instructions
474 )
476 return type_var_default_factory
477 elif values_type not in _defaultdict_allowed_default_types:
478 # a somewhat subjective set of types that have reasonable default values
479 allowed_msg = ', '.join([t.__name__ for t in _defaultdict_allowed_default_types.keys()])
480 raise PydanticSchemaGenerationError(
481 f'Unable to infer a default factory for values of type {values_source_type}.'
482 f' Only {allowed_msg} are supported, other types require an explicit default factory'
483 ' ' + instructions
484 )
485 return _defaultdict_allowed_default_types[values_type]
487 # Assume Annotated[..., Field(...)]
488 if typing_objects.is_annotated(values_type_origin):
489 field_info = next((v for v in get_args(values_source_type) if isinstance(v, FieldInfo)), None)
490 else:
491 field_info = None
492 if field_info and field_info.default_factory:
493 # Assume the default factory does not take any argument:
494 default_default_factory = cast(Callable[[], Any], field_info.default_factory)
495 else:
496 default_default_factory = infer_default()
497 return default_default_factory
500def validate_str_is_valid_iana_tz(value: Any, /) -> ZoneInfo:
501 if isinstance(value, ZoneInfo):
502 return value
503 try:
504 return ZoneInfo(value)
505 except (ZoneInfoNotFoundError, ValueError, TypeError, OSError):
506 raise PydanticCustomError('zoneinfo_str', 'invalid timezone: {value}', {'value': value})
509NUMERIC_VALIDATOR_LOOKUP: dict[str, Callable] = {
510 'gt': greater_than_validator,
511 'ge': greater_than_or_equal_validator,
512 'lt': less_than_validator,
513 'le': less_than_or_equal_validator,
514 'multiple_of': multiple_of_validator,
515 'min_length': min_length_validator,
516 'max_length': max_length_validator,
517 'max_digits': max_digits_validator,
518 'decimal_places': decimal_places_validator,
519}
521IpType: TypeAlias = IPv4Address | IPv6Address | IPv4Network | IPv6Network | IPv4Interface | IPv6Interface
523IP_VALIDATOR_LOOKUP: dict[type[IpType], Callable] = {
524 IPv4Address: ip_v4_address_validator,
525 IPv6Address: ip_v6_address_validator,
526 IPv4Network: ip_v4_network_validator,
527 IPv6Network: ip_v6_network_validator,
528 IPv4Interface: ip_v4_interface_validator,
529 IPv6Interface: ip_v6_interface_validator,
530}
532MAPPING_ORIGIN_MAP: dict[Any, Any] = {
533 typing.DefaultDict: collections.defaultdict, # noqa: UP006
534 collections.defaultdict: collections.defaultdict,
535 # this doesn't handle subclasses of these
536 typing.Mapping: dict,
537 typing.MutableMapping: dict,
538 # parametrized typing.{Mutable}Mapping creates one of these
539 collections.abc.Mapping: dict,
540 collections.abc.MutableMapping: dict,
541}