diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/pydantic/_internal/_validate_call.py')
-rw-r--r-- | .venv/lib/python3.12/site-packages/pydantic/_internal/_validate_call.py | 115 |
1 files changed, 115 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/pydantic/_internal/_validate_call.py b/.venv/lib/python3.12/site-packages/pydantic/_internal/_validate_call.py new file mode 100644 index 00000000..f04da826 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/pydantic/_internal/_validate_call.py @@ -0,0 +1,115 @@ +from __future__ import annotations as _annotations + +import functools +import inspect +from functools import partial +from typing import Any, Awaitable, Callable + +import pydantic_core + +from ..config import ConfigDict +from ..plugin._schema_validator import create_schema_validator +from ._config import ConfigWrapper +from ._generate_schema import GenerateSchema, ValidateCallSupportedTypes +from ._namespace_utils import MappingNamespace, NsResolver, ns_for_function + + +def extract_function_name(func: ValidateCallSupportedTypes) -> str: + """Extract the name of a `ValidateCallSupportedTypes` object.""" + return f'partial({func.func.__name__})' if isinstance(func, functools.partial) else func.__name__ + + +def extract_function_qualname(func: ValidateCallSupportedTypes) -> str: + """Extract the qualname of a `ValidateCallSupportedTypes` object.""" + return f'partial({func.func.__qualname__})' if isinstance(func, functools.partial) else func.__qualname__ + + +def update_wrapper_attributes(wrapped: ValidateCallSupportedTypes, wrapper: Callable[..., Any]): + """Update the `wrapper` function with the attributes of the `wrapped` function. Return the updated function.""" + if inspect.iscoroutinefunction(wrapped): + + @functools.wraps(wrapped) + async def wrapper_function(*args, **kwargs): # type: ignore + return await wrapper(*args, **kwargs) + else: + + @functools.wraps(wrapped) + def wrapper_function(*args, **kwargs): + return wrapper(*args, **kwargs) + + # We need to manually update this because `partial` object has no `__name__` and `__qualname__`. + wrapper_function.__name__ = extract_function_name(wrapped) + wrapper_function.__qualname__ = extract_function_qualname(wrapped) + wrapper_function.raw_function = wrapped # type: ignore + + return wrapper_function + + +class ValidateCallWrapper: + """This is a wrapper around a function that validates the arguments passed to it, and optionally the return value.""" + + __slots__ = ('__pydantic_validator__', '__return_pydantic_validator__') + + def __init__( + self, + function: ValidateCallSupportedTypes, + config: ConfigDict | None, + validate_return: bool, + parent_namespace: MappingNamespace | None, + ) -> None: + if isinstance(function, partial): + schema_type = function.func + module = function.func.__module__ + else: + schema_type = function + module = function.__module__ + qualname = extract_function_qualname(function) + + ns_resolver = NsResolver(namespaces_tuple=ns_for_function(schema_type, parent_namespace=parent_namespace)) + + config_wrapper = ConfigWrapper(config) + gen_schema = GenerateSchema(config_wrapper, ns_resolver) + schema = gen_schema.clean_schema(gen_schema.generate_schema(function)) + core_config = config_wrapper.core_config(title=qualname) + + self.__pydantic_validator__ = create_schema_validator( + schema, + schema_type, + module, + qualname, + 'validate_call', + core_config, + config_wrapper.plugin_settings, + ) + + if validate_return: + signature = inspect.signature(function) + return_type = signature.return_annotation if signature.return_annotation is not signature.empty else Any + gen_schema = GenerateSchema(config_wrapper, ns_resolver) + schema = gen_schema.clean_schema(gen_schema.generate_schema(return_type)) + validator = create_schema_validator( + schema, + schema_type, + module, + qualname, + 'validate_call', + core_config, + config_wrapper.plugin_settings, + ) + if inspect.iscoroutinefunction(function): + + async def return_val_wrapper(aw: Awaitable[Any]) -> None: + return validator.validate_python(await aw) + + self.__return_pydantic_validator__ = return_val_wrapper + else: + self.__return_pydantic_validator__ = validator.validate_python + else: + self.__return_pydantic_validator__ = None + + def __call__(self, *args: Any, **kwargs: Any) -> Any: + res = self.__pydantic_validator__.validate_python(pydantic_core.ArgsKwargs(args, kwargs)) + if self.__return_pydantic_validator__: + return self.__return_pydantic_validator__(res) + else: + return res |