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

69 statements  

« prev     ^ index     » next       coverage.py v7.10.0, created at 2025-07-26 11:49 +0000

1from __future__ import annotations as _annotations 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

2 

3import functools 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

4import inspect 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

5from collections.abc import Awaitable 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

6from functools import partial 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

7from typing import Any, Callable 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

8 

9import pydantic_core 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

10 

11from ..config import ConfigDict 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

12from ..plugin._schema_validator import create_schema_validator 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

13from ._config import ConfigWrapper 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

14from ._generate_schema import GenerateSchema, ValidateCallSupportedTypes 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

15from ._namespace_utils import MappingNamespace, NsResolver, ns_for_function 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

16 

17 

18def extract_function_name(func: ValidateCallSupportedTypes) -> str: 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

19 """Extract the name of a `ValidateCallSupportedTypes` object.""" 

20 return f'partial({func.func.__name__})' if isinstance(func, functools.partial) else func.__name__ 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

21 

22 

23def extract_function_qualname(func: ValidateCallSupportedTypes) -> str: 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

24 """Extract the qualname of a `ValidateCallSupportedTypes` object.""" 

25 return f'partial({func.func.__qualname__})' if isinstance(func, functools.partial) else func.__qualname__ 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

26 

27 

28def update_wrapper_attributes(wrapped: ValidateCallSupportedTypes, wrapper: Callable[..., Any]): 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

29 """Update the `wrapper` function with the attributes of the `wrapped` function. Return the updated function.""" 

30 if inspect.iscoroutinefunction(wrapped): 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

31 

32 @functools.wraps(wrapped) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

33 async def wrapper_function(*args, **kwargs): # type: ignore 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

34 return await wrapper(*args, **kwargs) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

35 else: 

36 

37 @functools.wraps(wrapped) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

38 def wrapper_function(*args, **kwargs): 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

39 return wrapper(*args, **kwargs) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

40 

41 # We need to manually update this because `partial` object has no `__name__` and `__qualname__`. 

42 wrapper_function.__name__ = extract_function_name(wrapped) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

43 wrapper_function.__qualname__ = extract_function_qualname(wrapped) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

44 wrapper_function.raw_function = wrapped # type: ignore 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

45 

46 return wrapper_function 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

47 

48 

49class ValidateCallWrapper: 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

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

51 

52 __slots__ = ( 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

53 'function', 

54 'validate_return', 

55 'schema_type', 

56 'module', 

57 'qualname', 

58 'ns_resolver', 

59 'config_wrapper', 

60 '__pydantic_complete__', 

61 '__pydantic_validator__', 

62 '__return_pydantic_validator__', 

63 ) 

64 

65 def __init__( 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

66 self, 

67 function: ValidateCallSupportedTypes, 

68 config: ConfigDict | None, 

69 validate_return: bool, 

70 parent_namespace: MappingNamespace | None, 

71 ) -> None: 

72 self.function = function 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

73 self.validate_return = validate_return 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

74 if isinstance(function, partial): 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

75 self.schema_type = function.func 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

76 self.module = function.func.__module__ 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

77 else: 

78 self.schema_type = function 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

79 self.module = function.__module__ 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

80 self.qualname = extract_function_qualname(function) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

81 

82 self.ns_resolver = NsResolver( 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

83 namespaces_tuple=ns_for_function(self.schema_type, parent_namespace=parent_namespace) 

84 ) 

85 self.config_wrapper = ConfigWrapper(config) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

86 if not self.config_wrapper.defer_build: 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

87 self._create_validators() 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

88 else: 

89 self.__pydantic_complete__ = False 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

90 

91 def _create_validators(self) -> None: 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

92 gen_schema = GenerateSchema(self.config_wrapper, self.ns_resolver) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

93 schema = gen_schema.clean_schema(gen_schema.generate_schema(self.function)) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

94 core_config = self.config_wrapper.core_config(title=self.qualname) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

95 

96 self.__pydantic_validator__ = create_schema_validator( 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

97 schema, 

98 self.schema_type, 

99 self.module, 

100 self.qualname, 

101 'validate_call', 

102 core_config, 

103 self.config_wrapper.plugin_settings, 

104 ) 

105 if self.validate_return: 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

106 signature = inspect.signature(self.function) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

107 return_type = signature.return_annotation if signature.return_annotation is not signature.empty else Any 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

108 gen_schema = GenerateSchema(self.config_wrapper, self.ns_resolver) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

109 schema = gen_schema.clean_schema(gen_schema.generate_schema(return_type)) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

110 validator = create_schema_validator( 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

111 schema, 

112 self.schema_type, 

113 self.module, 

114 self.qualname, 

115 'validate_call', 

116 core_config, 

117 self.config_wrapper.plugin_settings, 

118 ) 

119 if inspect.iscoroutinefunction(self.function): 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

120 

121 async def return_val_wrapper(aw: Awaitable[Any]) -> None: 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

122 return validator.validate_python(await aw) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

123 

124 self.__return_pydantic_validator__ = return_val_wrapper 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

125 else: 

126 self.__return_pydantic_validator__ = validator.validate_python 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

127 else: 

128 self.__return_pydantic_validator__ = None 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

129 

130 self.__pydantic_complete__ = True 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

131 

132 def __call__(self, *args: Any, **kwargs: Any) -> Any: 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

133 if not self.__pydantic_complete__: 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

134 self._create_validators() 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

135 

136 res = self.__pydantic_validator__.validate_python(pydantic_core.ArgsKwargs(args, kwargs)) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

137 if self.__return_pydantic_validator__: 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

138 return self.__return_pydantic_validator__(res) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO

139 else: 

140 return res 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNO