Coverage for pydantic/_internal/_validate_call.py: 100.00%
47 statements
« prev ^ index » next coverage.py v7.5.4, created at 2024-07-03 19:29 +0000
« prev ^ index » next coverage.py v7.5.4, created at 2024-07-03 19:29 +0000
1from __future__ import annotations as _annotations 1abcdefghijklmnopqrstuvwxyzMNOPQRSTUVABCDEFGHIJKL
3import inspect 1abcdefghijklmnopqrstuvwxyzMNOPQRSTUVABCDEFGHIJKL
4from functools import partial 1abcdefghijklmnopqrstuvwxyzMNOPQRSTUVABCDEFGHIJKL
5from typing import Any, Awaitable, Callable 1abcdefghijklmnopqrstuvwxyzMNOPQRSTUVABCDEFGHIJKL
7import pydantic_core 1abcdefghijklmnopqrstuvwxyzMNOPQRSTUVABCDEFGHIJKL
9from ..config import ConfigDict 1abcdefghijklmnopqrstuvwxyzMNOPQRSTUVABCDEFGHIJKL
10from ..plugin._schema_validator import create_schema_validator 1abcdefghijklmnopqrstuvwxyzMNOPQRSTUVABCDEFGHIJKL
11from . import _generate_schema, _typing_extra 1abcdefghijklmnopqrstuvwxyzMNOPQRSTUVABCDEFGHIJKL
12from ._config import ConfigWrapper 1abcdefghijklmnopqrstuvwxyzMNOPQRSTUVABCDEFGHIJKL
15class ValidateCallWrapper: 1abcdefghijklmnopqrstuvwxyzMNOPQRSTUVABCDEFGHIJKL
16 """This is a wrapper around a function that validates the arguments passed to it, and optionally the return value."""
18 __slots__ = ( 1abcdefghijklmnopqrstuvwxyzMNOPQRSTUVABCDEFGHIJKL
19 '__pydantic_validator__',
20 '__name__',
21 '__qualname__',
22 '__annotations__',
23 '__dict__', # required for __module__
24 )
26 def __init__( 1abcdefghijklmnopqrstuvwxyzMNOPQRSTUVABCDEFGHIJKL
27 self,
28 function: Callable[..., Any],
29 config: ConfigDict | None,
30 validate_return: bool,
31 namespace: dict[str, Any] | None,
32 ):
33 if isinstance(function, partial): 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
34 func = function.func 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
35 schema_type = func 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
36 self.__name__ = f'partial({func.__name__})' 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
37 self.__qualname__ = f'partial({func.__qualname__})' 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
38 self.__module__ = func.__module__ 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
39 else:
40 schema_type = function 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
41 self.__name__ = function.__name__ 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
42 self.__qualname__ = function.__qualname__ 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
43 self.__module__ = function.__module__ 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
45 global_ns = _typing_extra.add_module_globals(function, None) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
46 # TODO: this is a bit of a hack, we should probably have a better way to handle this
47 # specifically, we shouldn't be pumping the namespace full of type_params
48 # when we take namespace and type_params arguments in eval_type_backport
49 type_params = getattr(schema_type, '__type_params__', ()) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
50 namespace = { 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
51 **{param.__name__: param for param in type_params},
52 **(global_ns or {}),
53 **(namespace or {}),
54 }
55 config_wrapper = ConfigWrapper(config) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
56 gen_schema = _generate_schema.GenerateSchema(config_wrapper, namespace) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
57 schema = gen_schema.clean_schema(gen_schema.generate_schema(function)) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
58 core_config = config_wrapper.core_config(self) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
60 self.__pydantic_validator__ = create_schema_validator( 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
61 schema,
62 schema_type,
63 self.__module__,
64 self.__qualname__,
65 'validate_call',
66 core_config,
67 config_wrapper.plugin_settings,
68 )
70 if validate_return: 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
71 signature = inspect.signature(function) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
72 return_type = signature.return_annotation if signature.return_annotation is not signature.empty else Any 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
73 gen_schema = _generate_schema.GenerateSchema(config_wrapper, namespace) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
74 schema = gen_schema.clean_schema(gen_schema.generate_schema(return_type)) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
75 validator = create_schema_validator( 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
76 schema,
77 schema_type,
78 self.__module__,
79 self.__qualname__,
80 'validate_call',
81 core_config,
82 config_wrapper.plugin_settings,
83 )
84 if inspect.iscoroutinefunction(function): 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
86 async def return_val_wrapper(aw: Awaitable[Any]) -> None: 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
87 return validator.validate_python(await aw) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
89 self.__return_pydantic_validator__ = return_val_wrapper 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
90 else:
91 self.__return_pydantic_validator__ = validator.validate_python 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
92 else:
93 self.__return_pydantic_validator__ = None 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
95 def __call__(self, *args: Any, **kwargs: Any) -> Any: 1abcdefghijklmnopqrstuvwxyzMNOPQRSTUVABCDEFGHIJKL
96 res = self.__pydantic_validator__.validate_python(pydantic_core.ArgsKwargs(args, kwargs)) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
97 if self.__return_pydantic_validator__: 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
98 return self.__return_pydantic_validator__(res) 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL
99 return res 1abcdefghijklmnopqrstuvwxyzABCDEFGHIJKL