Coverage for pydantic/_internal/_validate_call.py: 100.00%

45 statements  

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

1from __future__ import annotations as _annotations 1abcdefghijklmnopqrstuvGHIJKLMNOwxyzABCDEF

2 

3import inspect 1abcdefghijklmnopqrstuvGHIJKLMNOwxyzABCDEF

4from functools import partial 1abcdefghijklmnopqrstuvGHIJKLMNOwxyzABCDEF

5from typing import Any, Awaitable, Callable 1abcdefghijklmnopqrstuvGHIJKLMNOwxyzABCDEF

6 

7import pydantic_core 1abcdefghijklmnopqrstuvGHIJKLMNOwxyzABCDEF

8 

9from ..config import ConfigDict 1abcdefghijklmnopqrstuvGHIJKLMNOwxyzABCDEF

10from ..plugin._schema_validator import create_schema_validator 1abcdefghijklmnopqrstuvGHIJKLMNOwxyzABCDEF

11from . import _generate_schema, _typing_extra 1abcdefghijklmnopqrstuvGHIJKLMNOwxyzABCDEF

12from ._config import ConfigWrapper 1abcdefghijklmnopqrstuvGHIJKLMNOwxyzABCDEF

13 

14 

15class ValidateCallWrapper: 1abcdefghijklmnopqrstuvGHIJKLMNOwxyzABCDEF

16 """This is a wrapper around a function that validates the arguments passed to it, and optionally the return value.""" 

17 

18 __slots__ = ( 1abcdefghijklmnopqrstuvGHIJKLMNOwxyzABCDEF

19 '__pydantic_validator__', 

20 '__name__', 

21 '__qualname__', 

22 '__annotations__', 

23 '__dict__', # required for __module__ 

24 ) 

25 

26 def __init__(self, function: Callable[..., Any], config: ConfigDict | None, validate_return: bool): 1abcdefghijklmnopqrstuvGHIJKLMNOwxyzABCDEF

27 if isinstance(function, partial): 1abcdefghijklmnopqrstuvwxyzABCDEF

28 func = function.func 1abcdefghijklmnopqrstuvwxyzABCDEF

29 schema_type = func 1abcdefghijklmnopqrstuvwxyzABCDEF

30 self.__name__ = f'partial({func.__name__})' 1abcdefghijklmnopqrstuvwxyzABCDEF

31 self.__qualname__ = f'partial({func.__qualname__})' 1abcdefghijklmnopqrstuvwxyzABCDEF

32 self.__module__ = func.__module__ 1abcdefghijklmnopqrstuvwxyzABCDEF

33 else: 

34 schema_type = function 1abcdefghijklmnopqrstuvwxyzABCDEF

35 self.__name__ = function.__name__ 1abcdefghijklmnopqrstuvwxyzABCDEF

36 self.__qualname__ = function.__qualname__ 1abcdefghijklmnopqrstuvwxyzABCDEF

37 self.__module__ = function.__module__ 1abcdefghijklmnopqrstuvwxyzABCDEF

38 

39 namespace = _typing_extra.add_module_globals(function, None) 1abcdefghijklmnopqrstuvwxyzABCDEF

40 config_wrapper = ConfigWrapper(config) 1abcdefghijklmnopqrstuvwxyzABCDEF

41 gen_schema = _generate_schema.GenerateSchema(config_wrapper, namespace) 1abcdefghijklmnopqrstuvwxyzABCDEF

42 schema = gen_schema.clean_schema(gen_schema.generate_schema(function)) 1abcdefghijklmnopqrstuvwxyzABCDEF

43 core_config = config_wrapper.core_config(self) 1abcdefghijklmnopqrstuvwxyzABCDEF

44 

45 self.__pydantic_validator__ = create_schema_validator( 1abcdefghijklmnopqrstuvwxyzABCDEF

46 schema, 

47 schema_type, 

48 self.__module__, 

49 self.__qualname__, 

50 'validate_call', 

51 core_config, 

52 config_wrapper.plugin_settings, 

53 ) 

54 

55 if validate_return: 1abcdefghijklmnopqrstuvwxyzABCDEF

56 signature = inspect.signature(function) 1abcdefghijklmnopqrstuvwxyzABCDEF

57 return_type = signature.return_annotation if signature.return_annotation is not signature.empty else Any 1abcdefghijklmnopqrstuvwxyzABCDEF

58 gen_schema = _generate_schema.GenerateSchema(config_wrapper, namespace) 1abcdefghijklmnopqrstuvwxyzABCDEF

59 schema = gen_schema.clean_schema(gen_schema.generate_schema(return_type)) 1abcdefghijklmnopqrstuvwxyzABCDEF

60 validator = create_schema_validator( 1abcdefghijklmnopqrstuvwxyzABCDEF

61 schema, 

62 schema_type, 

63 self.__module__, 

64 self.__qualname__, 

65 'validate_call', 

66 core_config, 

67 config_wrapper.plugin_settings, 

68 ) 

69 if inspect.iscoroutinefunction(function): 1abcdefghijklmnopqrstuvwxyzABCDEF

70 

71 async def return_val_wrapper(aw: Awaitable[Any]) -> None: 1abcdefghijklmnopqrstuvwxyzABCDEF

72 return validator.validate_python(await aw) 1abcdefghijklmnopqrstuvwxyzABCDEF

73 

74 self.__return_pydantic_validator__ = return_val_wrapper 1abcdefghijklmnopqrstuvwxyzABCDEF

75 else: 

76 self.__return_pydantic_validator__ = validator.validate_python 1abcdefghijklmnopqrstuvwxyzABCDEF

77 else: 

78 self.__return_pydantic_validator__ = None 1abcdefghijklmnopqrstuvwxyzABCDEF

79 

80 def __call__(self, *args: Any, **kwargs: Any) -> Any: 1abcdefghijklmnopqrstuvGHIJKLMNOwxyzABCDEF

81 res = self.__pydantic_validator__.validate_python(pydantic_core.ArgsKwargs(args, kwargs)) 1abcdefghijklmnopqrstuvwxyzABCDEF

82 if self.__return_pydantic_validator__: 1abcdefghijklmnopqrstuvwxyzABCDEF

83 return self.__return_pydantic_validator__(res) 1abcdefghijklmnopqrstuvwxyzABCDEF

84 return res 1abcdefghijklmnopqrstuvwxyzABCDEF