class_validators.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361
  1. import warnings
  2. from collections import ChainMap
  3. from functools import partial, partialmethod, wraps
  4. from itertools import chain
  5. from types import FunctionType
  6. from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, List, Optional, Set, Tuple, Type, Union, overload
  7. from pydantic.v1.errors import ConfigError
  8. from pydantic.v1.typing import AnyCallable
  9. from pydantic.v1.utils import ROOT_KEY, in_ipython
  10. if TYPE_CHECKING:
  11. from pydantic.v1.typing import AnyClassMethod
  12. class Validator:
  13. __slots__ = 'func', 'pre', 'each_item', 'always', 'check_fields', 'skip_on_failure'
  14. def __init__(
  15. self,
  16. func: AnyCallable,
  17. pre: bool = False,
  18. each_item: bool = False,
  19. always: bool = False,
  20. check_fields: bool = False,
  21. skip_on_failure: bool = False,
  22. ):
  23. self.func = func
  24. self.pre = pre
  25. self.each_item = each_item
  26. self.always = always
  27. self.check_fields = check_fields
  28. self.skip_on_failure = skip_on_failure
  29. if TYPE_CHECKING:
  30. from inspect import Signature
  31. from pydantic.v1.config import BaseConfig
  32. from pydantic.v1.fields import ModelField
  33. from pydantic.v1.types import ModelOrDc
  34. ValidatorCallable = Callable[[Optional[ModelOrDc], Any, Dict[str, Any], ModelField, Type[BaseConfig]], Any]
  35. ValidatorsList = List[ValidatorCallable]
  36. ValidatorListDict = Dict[str, List[Validator]]
  37. _FUNCS: Set[str] = set()
  38. VALIDATOR_CONFIG_KEY = '__validator_config__'
  39. ROOT_VALIDATOR_CONFIG_KEY = '__root_validator_config__'
  40. def validator(
  41. *fields: str,
  42. pre: bool = False,
  43. each_item: bool = False,
  44. always: bool = False,
  45. check_fields: bool = True,
  46. whole: Optional[bool] = None,
  47. allow_reuse: bool = False,
  48. ) -> Callable[[AnyCallable], 'AnyClassMethod']:
  49. """
  50. Decorate methods on the class indicating that they should be used to validate fields
  51. :param fields: which field(s) the method should be called on
  52. :param pre: whether or not this validator should be called before the standard validators (else after)
  53. :param each_item: for complex objects (sets, lists etc.) whether to validate individual elements rather than the
  54. whole object
  55. :param always: whether this method and other validators should be called even if the value is missing
  56. :param check_fields: whether to check that the fields actually exist on the model
  57. :param allow_reuse: whether to track and raise an error if another validator refers to the decorated function
  58. """
  59. if not fields:
  60. raise ConfigError('validator with no fields specified')
  61. elif isinstance(fields[0], FunctionType):
  62. raise ConfigError(
  63. "validators should be used with fields and keyword arguments, not bare. " # noqa: Q000
  64. "E.g. usage should be `@validator('<field_name>', ...)`"
  65. )
  66. elif not all(isinstance(field, str) for field in fields):
  67. raise ConfigError(
  68. "validator fields should be passed as separate string args. " # noqa: Q000
  69. "E.g. usage should be `@validator('<field_name_1>', '<field_name_2>', ...)`"
  70. )
  71. if whole is not None:
  72. warnings.warn(
  73. 'The "whole" keyword argument is deprecated, use "each_item" (inverse meaning, default False) instead',
  74. DeprecationWarning,
  75. )
  76. assert each_item is False, '"each_item" and "whole" conflict, remove "whole"'
  77. each_item = not whole
  78. def dec(f: AnyCallable) -> 'AnyClassMethod':
  79. f_cls = _prepare_validator(f, allow_reuse)
  80. setattr(
  81. f_cls,
  82. VALIDATOR_CONFIG_KEY,
  83. (
  84. fields,
  85. Validator(func=f_cls.__func__, pre=pre, each_item=each_item, always=always, check_fields=check_fields),
  86. ),
  87. )
  88. return f_cls
  89. return dec
  90. @overload
  91. def root_validator(_func: AnyCallable) -> 'AnyClassMethod':
  92. ...
  93. @overload
  94. def root_validator(
  95. *, pre: bool = False, allow_reuse: bool = False, skip_on_failure: bool = False
  96. ) -> Callable[[AnyCallable], 'AnyClassMethod']:
  97. ...
  98. def root_validator(
  99. _func: Optional[AnyCallable] = None, *, pre: bool = False, allow_reuse: bool = False, skip_on_failure: bool = False
  100. ) -> Union['AnyClassMethod', Callable[[AnyCallable], 'AnyClassMethod']]:
  101. """
  102. Decorate methods on a model indicating that they should be used to validate (and perhaps modify) data either
  103. before or after standard model parsing/validation is performed.
  104. """
  105. if _func:
  106. f_cls = _prepare_validator(_func, allow_reuse)
  107. setattr(
  108. f_cls, ROOT_VALIDATOR_CONFIG_KEY, Validator(func=f_cls.__func__, pre=pre, skip_on_failure=skip_on_failure)
  109. )
  110. return f_cls
  111. def dec(f: AnyCallable) -> 'AnyClassMethod':
  112. f_cls = _prepare_validator(f, allow_reuse)
  113. setattr(
  114. f_cls, ROOT_VALIDATOR_CONFIG_KEY, Validator(func=f_cls.__func__, pre=pre, skip_on_failure=skip_on_failure)
  115. )
  116. return f_cls
  117. return dec
  118. def _prepare_validator(function: AnyCallable, allow_reuse: bool) -> 'AnyClassMethod':
  119. """
  120. Avoid validators with duplicated names since without this, validators can be overwritten silently
  121. which generally isn't the intended behaviour, don't run in ipython (see #312) or if allow_reuse is False.
  122. """
  123. f_cls = function if isinstance(function, classmethod) else classmethod(function)
  124. if not in_ipython() and not allow_reuse:
  125. ref = (
  126. getattr(f_cls.__func__, '__module__', '<No __module__>')
  127. + '.'
  128. + getattr(f_cls.__func__, '__qualname__', f'<No __qualname__: id:{id(f_cls.__func__)}>')
  129. )
  130. if ref in _FUNCS:
  131. raise ConfigError(f'duplicate validator function "{ref}"; if this is intended, set `allow_reuse=True`')
  132. _FUNCS.add(ref)
  133. return f_cls
  134. class ValidatorGroup:
  135. def __init__(self, validators: 'ValidatorListDict') -> None:
  136. self.validators = validators
  137. self.used_validators = {'*'}
  138. def get_validators(self, name: str) -> Optional[Dict[str, Validator]]:
  139. self.used_validators.add(name)
  140. validators = self.validators.get(name, [])
  141. if name != ROOT_KEY:
  142. validators += self.validators.get('*', [])
  143. if validators:
  144. return {getattr(v.func, '__name__', f'<No __name__: id:{id(v.func)}>'): v for v in validators}
  145. else:
  146. return None
  147. def check_for_unused(self) -> None:
  148. unused_validators = set(
  149. chain.from_iterable(
  150. (
  151. getattr(v.func, '__name__', f'<No __name__: id:{id(v.func)}>')
  152. for v in self.validators[f]
  153. if v.check_fields
  154. )
  155. for f in (self.validators.keys() - self.used_validators)
  156. )
  157. )
  158. if unused_validators:
  159. fn = ', '.join(unused_validators)
  160. raise ConfigError(
  161. f"Validators defined with incorrect fields: {fn} " # noqa: Q000
  162. f"(use check_fields=False if you're inheriting from the model and intended this)"
  163. )
  164. def extract_validators(namespace: Dict[str, Any]) -> Dict[str, List[Validator]]:
  165. validators: Dict[str, List[Validator]] = {}
  166. for var_name, value in namespace.items():
  167. validator_config = getattr(value, VALIDATOR_CONFIG_KEY, None)
  168. if validator_config:
  169. fields, v = validator_config
  170. for field in fields:
  171. if field in validators:
  172. validators[field].append(v)
  173. else:
  174. validators[field] = [v]
  175. return validators
  176. def extract_root_validators(namespace: Dict[str, Any]) -> Tuple[List[AnyCallable], List[Tuple[bool, AnyCallable]]]:
  177. from inspect import signature
  178. pre_validators: List[AnyCallable] = []
  179. post_validators: List[Tuple[bool, AnyCallable]] = []
  180. for name, value in namespace.items():
  181. validator_config: Optional[Validator] = getattr(value, ROOT_VALIDATOR_CONFIG_KEY, None)
  182. if validator_config:
  183. sig = signature(validator_config.func)
  184. args = list(sig.parameters.keys())
  185. if args[0] == 'self':
  186. raise ConfigError(
  187. f'Invalid signature for root validator {name}: {sig}, "self" not permitted as first argument, '
  188. f'should be: (cls, values).'
  189. )
  190. if len(args) != 2:
  191. raise ConfigError(f'Invalid signature for root validator {name}: {sig}, should be: (cls, values).')
  192. # check function signature
  193. if validator_config.pre:
  194. pre_validators.append(validator_config.func)
  195. else:
  196. post_validators.append((validator_config.skip_on_failure, validator_config.func))
  197. return pre_validators, post_validators
  198. def inherit_validators(base_validators: 'ValidatorListDict', validators: 'ValidatorListDict') -> 'ValidatorListDict':
  199. for field, field_validators in base_validators.items():
  200. if field not in validators:
  201. validators[field] = []
  202. validators[field] += field_validators
  203. return validators
  204. def make_generic_validator(validator: AnyCallable) -> 'ValidatorCallable':
  205. """
  206. Make a generic function which calls a validator with the right arguments.
  207. Unfortunately other approaches (eg. return a partial of a function that builds the arguments) is slow,
  208. hence this laborious way of doing things.
  209. It's done like this so validators don't all need **kwargs in their signature, eg. any combination of
  210. the arguments "values", "fields" and/or "config" are permitted.
  211. """
  212. from inspect import signature
  213. if not isinstance(validator, (partial, partialmethod)):
  214. # This should be the default case, so overhead is reduced
  215. sig = signature(validator)
  216. args = list(sig.parameters.keys())
  217. else:
  218. # Fix the generated argument lists of partial methods
  219. sig = signature(validator.func)
  220. args = [
  221. k
  222. for k in signature(validator.func).parameters.keys()
  223. if k not in validator.args | validator.keywords.keys()
  224. ]
  225. first_arg = args.pop(0)
  226. if first_arg == 'self':
  227. raise ConfigError(
  228. f'Invalid signature for validator {validator}: {sig}, "self" not permitted as first argument, '
  229. f'should be: (cls, value, values, config, field), "values", "config" and "field" are all optional.'
  230. )
  231. elif first_arg == 'cls':
  232. # assume the second argument is value
  233. return wraps(validator)(_generic_validator_cls(validator, sig, set(args[1:])))
  234. else:
  235. # assume the first argument was value which has already been removed
  236. return wraps(validator)(_generic_validator_basic(validator, sig, set(args)))
  237. def prep_validators(v_funcs: Iterable[AnyCallable]) -> 'ValidatorsList':
  238. return [make_generic_validator(f) for f in v_funcs if f]
  239. all_kwargs = {'values', 'field', 'config'}
  240. def _generic_validator_cls(validator: AnyCallable, sig: 'Signature', args: Set[str]) -> 'ValidatorCallable':
  241. # assume the first argument is value
  242. has_kwargs = False
  243. if 'kwargs' in args:
  244. has_kwargs = True
  245. args -= {'kwargs'}
  246. if not args.issubset(all_kwargs):
  247. raise ConfigError(
  248. f'Invalid signature for validator {validator}: {sig}, should be: '
  249. f'(cls, value, values, config, field), "values", "config" and "field" are all optional.'
  250. )
  251. if has_kwargs:
  252. return lambda cls, v, values, field, config: validator(cls, v, values=values, field=field, config=config)
  253. elif args == set():
  254. return lambda cls, v, values, field, config: validator(cls, v)
  255. elif args == {'values'}:
  256. return lambda cls, v, values, field, config: validator(cls, v, values=values)
  257. elif args == {'field'}:
  258. return lambda cls, v, values, field, config: validator(cls, v, field=field)
  259. elif args == {'config'}:
  260. return lambda cls, v, values, field, config: validator(cls, v, config=config)
  261. elif args == {'values', 'field'}:
  262. return lambda cls, v, values, field, config: validator(cls, v, values=values, field=field)
  263. elif args == {'values', 'config'}:
  264. return lambda cls, v, values, field, config: validator(cls, v, values=values, config=config)
  265. elif args == {'field', 'config'}:
  266. return lambda cls, v, values, field, config: validator(cls, v, field=field, config=config)
  267. else:
  268. # args == {'values', 'field', 'config'}
  269. return lambda cls, v, values, field, config: validator(cls, v, values=values, field=field, config=config)
  270. def _generic_validator_basic(validator: AnyCallable, sig: 'Signature', args: Set[str]) -> 'ValidatorCallable':
  271. has_kwargs = False
  272. if 'kwargs' in args:
  273. has_kwargs = True
  274. args -= {'kwargs'}
  275. if not args.issubset(all_kwargs):
  276. raise ConfigError(
  277. f'Invalid signature for validator {validator}: {sig}, should be: '
  278. f'(value, values, config, field), "values", "config" and "field" are all optional.'
  279. )
  280. if has_kwargs:
  281. return lambda cls, v, values, field, config: validator(v, values=values, field=field, config=config)
  282. elif args == set():
  283. return lambda cls, v, values, field, config: validator(v)
  284. elif args == {'values'}:
  285. return lambda cls, v, values, field, config: validator(v, values=values)
  286. elif args == {'field'}:
  287. return lambda cls, v, values, field, config: validator(v, field=field)
  288. elif args == {'config'}:
  289. return lambda cls, v, values, field, config: validator(v, config=config)
  290. elif args == {'values', 'field'}:
  291. return lambda cls, v, values, field, config: validator(v, values=values, field=field)
  292. elif args == {'values', 'config'}:
  293. return lambda cls, v, values, field, config: validator(v, values=values, config=config)
  294. elif args == {'field', 'config'}:
  295. return lambda cls, v, values, field, config: validator(v, field=field, config=config)
  296. else:
  297. # args == {'values', 'field', 'config'}
  298. return lambda cls, v, values, field, config: validator(v, values=values, field=field, config=config)
  299. def gather_all_validators(type_: 'ModelOrDc') -> Dict[str, 'AnyClassMethod']:
  300. all_attributes = ChainMap(*[cls.__dict__ for cls in type_.__mro__]) # type: ignore[arg-type,var-annotated]
  301. return {
  302. k: v
  303. for k, v in all_attributes.items()
  304. if hasattr(v, VALIDATOR_CONFIG_KEY) or hasattr(v, ROOT_VALIDATOR_CONFIG_KEY)
  305. }