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

1from __future__ import annotations as _annotations 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

2 

3import functools 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

4import inspect 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

5from collections.abc import Awaitable 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

6from functools import partial 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

7from typing import Any, Callable 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

8 

9import pydantic_core 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

10 

11from ..config import ConfigDict 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

12from ..plugin._schema_validator import create_schema_validator 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

13from ._config import ConfigWrapper 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

14from ._generate_schema import GenerateSchema, ValidateCallSupportedTypes 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

15from ._namespace_utils import MappingNamespace, NsResolver, ns_for_function 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

16 

17 

18def extract_function_name(func: ValidateCallSupportedTypes) -> str: 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

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

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

21 

22 

23def extract_function_qualname(func: ValidateCallSupportedTypes) -> str: 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

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

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

26 

27 

28def update_wrapper_attributes(wrapped: ValidateCallSupportedTypes, wrapper: Callable[..., Any]): 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

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

30 if inspect.iscoroutinefunction(wrapped): 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

31 

32 @functools.wraps(wrapped) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

33 async def wrapper_function(*args, **kwargs): # type: ignore 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

34 return await wrapper(*args, **kwargs) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

35 else: 

36 

37 @functools.wraps(wrapped) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

38 def wrapper_function(*args, **kwargs): 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

39 return wrapper(*args, **kwargs) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

40 

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

42 wrapper_function.__name__ = extract_function_name(wrapped) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

43 wrapper_function.__qualname__ = extract_function_qualname(wrapped) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

44 wrapper_function.raw_function = wrapped # type: ignore 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

45 

46 return wrapper_function 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

47 

48 

49class ValidateCallWrapper: 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

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

51 

52 __slots__ = ('__pydantic_validator__', '__return_pydantic_validator__') 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

53 

54 def __init__( 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

55 self, 

56 function: ValidateCallSupportedTypes, 

57 config: ConfigDict | None, 

58 validate_return: bool, 

59 parent_namespace: MappingNamespace | None, 

60 ) -> None: 

61 if isinstance(function, partial): 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

62 schema_type = function.func 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

63 module = function.func.__module__ 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

64 else: 

65 schema_type = function 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

66 module = function.__module__ 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

67 qualname = extract_function_qualname(function) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

68 

69 ns_resolver = NsResolver(namespaces_tuple=ns_for_function(schema_type, parent_namespace=parent_namespace)) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

70 

71 config_wrapper = ConfigWrapper(config) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

72 gen_schema = GenerateSchema(config_wrapper, ns_resolver) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

73 schema = gen_schema.clean_schema(gen_schema.generate_schema(function)) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

74 core_config = config_wrapper.core_config(title=qualname) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

75 

76 self.__pydantic_validator__ = create_schema_validator( 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

77 schema, 

78 schema_type, 

79 module, 

80 qualname, 

81 'validate_call', 

82 core_config, 

83 config_wrapper.plugin_settings, 

84 ) 

85 

86 if validate_return: 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

87 signature = inspect.signature(function) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

88 return_type = signature.return_annotation if signature.return_annotation is not signature.empty else Any 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

89 gen_schema = GenerateSchema(config_wrapper, ns_resolver) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

90 schema = gen_schema.clean_schema(gen_schema.generate_schema(return_type)) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

91 validator = create_schema_validator( 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

92 schema, 

93 schema_type, 

94 module, 

95 qualname, 

96 'validate_call', 

97 core_config, 

98 config_wrapper.plugin_settings, 

99 ) 

100 if inspect.iscoroutinefunction(function): 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

101 

102 async def return_val_wrapper(aw: Awaitable[Any]) -> None: 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

103 return validator.validate_python(await aw) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

104 

105 self.__return_pydantic_validator__ = return_val_wrapper 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

106 else: 

107 self.__return_pydantic_validator__ = validator.validate_python 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

108 else: 

109 self.__return_pydantic_validator__ = None 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

110 

111 def __call__(self, *args: Any, **kwargs: Any) -> Any: 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

112 res = self.__pydantic_validator__.validate_python(pydantic_core.ArgsKwargs(args, kwargs)) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

113 if self.__return_pydantic_validator__: 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

114 return self.__return_pydantic_validator__(res) 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF

115 else: 

116 return res 32 ctx1abcdefghijklmnopqrstuvwxyzABCDEF