aboutsummaryrefslogtreecommitdiff
path: root/.venv/lib/python3.12/site-packages/pydantic/_internal/_validate_call.py
diff options
context:
space:
mode:
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.py115
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