Загрузить файлы в «venv/Lib/site-packages/pydantic/plugin»

venv/Lib/site-packages/pydantic/plugin/__init__.py

@@ -0,0 +1,193 @@
+"""!!! abstract "Usage Documentation"
+    [Build a Plugin](../concepts/plugins.md#build-a-plugin)
+
+Plugin interface for Pydantic plugins, and related types.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Callable, Literal, NamedTuple
+
+from pydantic_core import CoreConfig, CoreSchema, ValidationError
+from typing_extensions import Protocol, TypeAlias
+
+from pydantic.config import ExtraValues
+
+__all__ = (
+    'PydanticPluginProtocol',
+    'BaseValidateHandlerProtocol',
+    'ValidatePythonHandlerProtocol',
+    'ValidateJsonHandlerProtocol',
+    'ValidateStringsHandlerProtocol',
+    'NewSchemaReturns',
+    'SchemaTypePath',
+    'SchemaKind',
+)
+
+NewSchemaReturns: TypeAlias = 'tuple[ValidatePythonHandlerProtocol | None, ValidateJsonHandlerProtocol | None, ValidateStringsHandlerProtocol | None]'
+
+
+class SchemaTypePath(NamedTuple):
+    """Path defining where `schema_type` was defined, or where `TypeAdapter` was called."""
+
+    module: str
+    name: str
+
+
+SchemaKind: TypeAlias = Literal['BaseModel', 'TypeAdapter', 'dataclass', 'create_model', 'validate_call']
+
+
+class PydanticPluginProtocol(Protocol):
+    """Protocol defining the interface for Pydantic plugins."""
+
+    def new_schema_validator(
+        self,
+        schema: CoreSchema,
+        schema_type: Any,
+        schema_type_path: SchemaTypePath,
+        schema_kind: SchemaKind,
+        config: CoreConfig | None,
+        plugin_settings: dict[str, object],
+    ) -> tuple[
+        ValidatePythonHandlerProtocol | None, ValidateJsonHandlerProtocol | None, ValidateStringsHandlerProtocol | None
+    ]:
+        """This method is called for each plugin every time a new [`SchemaValidator`][pydantic_core.SchemaValidator]
+        is created.
+
+        It should return an event handler for each of the three validation methods, or `None` if the plugin does not
+        implement that method.
+
+        Args:
+            schema: The schema to validate against.
+            schema_type: The original type which the schema was created from, e.g. the model class.
+            schema_type_path: Path defining where `schema_type` was defined, or where `TypeAdapter` was called.
+            schema_kind: The kind of schema to validate against.
+            config: The config to use for validation.
+            plugin_settings: Any plugin settings.
+
+        Returns:
+            A tuple of optional event handlers for each of the three validation methods -
+                `validate_python`, `validate_json`, `validate_strings`.
+        """
+        raise NotImplementedError('Pydantic plugins should implement `new_schema_validator`.')
+
+
+class BaseValidateHandlerProtocol(Protocol):
+    """Base class for plugin callbacks protocols.
+
+    You shouldn't implement this protocol directly, instead use one of the subclasses with adds the correctly
+    typed `on_error` method.
+    """
+
+    on_enter: Callable[..., None]
+    """`on_enter` is changed to be more specific on all subclasses"""
+
+    def on_success(self, result: Any) -> None:
+        """Callback to be notified of successful validation.
+
+        Args:
+            result: The result of the validation.
+        """
+        return
+
+    def on_error(self, error: ValidationError) -> None:
+        """Callback to be notified of validation errors.
+
+        Args:
+            error: The validation error.
+        """
+        return
+
+    def on_exception(self, exception: Exception) -> None:
+        """Callback to be notified of validation exceptions.
+
+        Args:
+            exception: The exception raised during validation.
+        """
+        return
+
+
+class ValidatePythonHandlerProtocol(BaseValidateHandlerProtocol, Protocol):
+    """Event handler for `SchemaValidator.validate_python`."""
+
+    def on_enter(
+        self,
+        input: Any,
+        *,
+        strict: bool | None = None,
+        extra: ExtraValues | None = None,
+        from_attributes: bool | None = None,
+        context: Any | None = None,
+        self_instance: Any | None = None,
+        by_alias: bool | None = None,
+        by_name: bool | None = None,
+    ) -> None:
+        """Callback to be notified of validation start, and create an instance of the event handler.
+
+        Args:
+            input: The input to be validated.
+            strict: Whether to validate the object in strict mode.
+            extra: Whether to ignore, allow, or forbid extra data during model validation.
+            from_attributes: Whether to validate objects as inputs by extracting attributes.
+            context: The context to use for validation, this is passed to functional validators.
+            self_instance: An instance of a model to set attributes on from validation, this is used when running
+                validation from the `__init__` method of a model.
+            by_alias: Whether to use the field's alias to match the input data to an attribute.
+            by_name: Whether to use the field's name to match the input data to an attribute.
+        """
+
+
+class ValidateJsonHandlerProtocol(BaseValidateHandlerProtocol, Protocol):
+    """Event handler for `SchemaValidator.validate_json`."""
+
+    def on_enter(
+        self,
+        input: str | bytes | bytearray,
+        *,
+        strict: bool | None = None,
+        extra: ExtraValues | None = None,
+        context: Any | None = None,
+        self_instance: Any | None = None,
+        by_alias: bool | None = None,
+        by_name: bool | None = None,
+    ) -> None:
+        """Callback to be notified of validation start, and create an instance of the event handler.
+
+        Args:
+            input: The JSON data to be validated.
+            strict: Whether to validate the object in strict mode.
+            extra: Whether to ignore, allow, or forbid extra data during model validation.
+            context: The context to use for validation, this is passed to functional validators.
+            self_instance: An instance of a model to set attributes on from validation, this is used when running
+                validation from the `__init__` method of a model.
+            by_alias: Whether to use the field's alias to match the input data to an attribute.
+            by_name: Whether to use the field's name to match the input data to an attribute.
+        """
+
+
+StringInput: TypeAlias = 'dict[str, StringInput]'
+
+
+class ValidateStringsHandlerProtocol(BaseValidateHandlerProtocol, Protocol):
+    """Event handler for `SchemaValidator.validate_strings`."""
+
+    def on_enter(
+        self,
+        input: StringInput,
+        *,
+        strict: bool | None = None,
+        extra: ExtraValues | None = None,
+        context: Any | None = None,
+        by_alias: bool | None = None,
+        by_name: bool | None = None,
+    ) -> None:
+        """Callback to be notified of validation start, and create an instance of the event handler.
+
+        Args:
+            input: The string data to be validated.
+            strict: Whether to validate the object in strict mode.
+            extra: Whether to ignore, allow, or forbid extra data during model validation.
+            context: The context to use for validation, this is passed to functional validators.
+            by_alias: Whether to use the field's alias to match the input data to an attribute.
+            by_name: Whether to use the field's name to match the input data to an attribute.
+        """

venv/Lib/site-packages/pydantic/plugin/_loader.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import importlib.metadata as importlib_metadata
+import os
+import warnings
+from collections.abc import Iterable
+from typing import TYPE_CHECKING, Final
+
+if TYPE_CHECKING:
+    from . import PydanticPluginProtocol
+
+
+PYDANTIC_ENTRY_POINT_GROUP: Final[str] = 'pydantic'
+
+# cache of plugins
+_plugins: dict[str, PydanticPluginProtocol] | None = None
+# return no plugins while loading plugins to avoid recursion and errors while importing plugins
+# this means that if plugins use pydantic
+_loading_plugins: bool = False
+
+
+def get_plugins() -> Iterable[PydanticPluginProtocol]:
+    """Load plugins for Pydantic.
+
+    Inspired by: https://github.com/pytest-dev/pluggy/blob/1.3.0/src/pluggy/_manager.py#L376-L402
+    """
+    disabled_plugins = os.getenv('PYDANTIC_DISABLE_PLUGINS')
+    global _plugins, _loading_plugins
+    if _loading_plugins:
+        # this happens when plugins themselves use pydantic, we return no plugins
+        return ()
+    elif disabled_plugins in ('__all__', '1', 'true'):
+        return ()
+    elif _plugins is None:
+        _plugins = {}
+        # set _loading_plugins so any plugins that use pydantic don't themselves use plugins
+        _loading_plugins = True
+        try:
+            for dist in importlib_metadata.distributions():
+                for entry_point in dist.entry_points:
+                    if entry_point.group != PYDANTIC_ENTRY_POINT_GROUP:
+                        continue
+                    if entry_point.value in _plugins:
+                        continue
+                    if disabled_plugins is not None and entry_point.name in disabled_plugins.split(','):
+                        continue
+                    try:
+                        _plugins[entry_point.value] = entry_point.load()
+                    except (ImportError, AttributeError) as e:
+                        warnings.warn(
+                            f'{e.__class__.__name__} while loading the `{entry_point.name}` Pydantic plugin, '
+                            f'this plugin will not be installed.\n\n{e!r}',
+                            stacklevel=2,
+                        )
+        finally:
+            _loading_plugins = False
+
+    return _plugins.values()

venv/Lib/site-packages/pydantic/plugin/_schema_validator.py

@@ -0,0 +1,143 @@
+"""Pluggable schema validator for pydantic."""
+
+from __future__ import annotations
+
+import functools
+from collections.abc import Iterable
+from typing import TYPE_CHECKING, Any, Callable, Literal, TypeVar
+
+from pydantic_core import CoreConfig, CoreSchema, SchemaValidator, ValidationError
+from typing_extensions import ParamSpec
+
+if TYPE_CHECKING:
+    from . import BaseValidateHandlerProtocol, PydanticPluginProtocol, SchemaKind, SchemaTypePath
+
+
+P = ParamSpec('P')
+R = TypeVar('R')
+Event = Literal['on_validate_python', 'on_validate_json', 'on_validate_strings']
+events: list[Event] = list(Event.__args__)  # type: ignore
+
+
+def create_schema_validator(
+    schema: CoreSchema,
+    schema_type: Any,
+    schema_type_module: str,
+    schema_type_name: str,
+    schema_kind: SchemaKind,
+    config: CoreConfig | None = None,
+    plugin_settings: dict[str, Any] | None = None,
+    _use_prebuilt: bool = True,
+) -> SchemaValidator | PluggableSchemaValidator:
+    """Create a `SchemaValidator` or `PluggableSchemaValidator` if plugins are installed.
+
+    Returns:
+        If plugins are installed then return `PluggableSchemaValidator`, otherwise return `SchemaValidator`.
+    """
+    from . import SchemaTypePath
+    from ._loader import get_plugins
+
+    plugins = get_plugins()
+    if plugins:
+        return PluggableSchemaValidator(
+            schema,
+            schema_type,
+            SchemaTypePath(schema_type_module, schema_type_name),
+            schema_kind,
+            config,
+            plugins,
+            plugin_settings or {},
+            _use_prebuilt=_use_prebuilt,
+        )
+    else:
+        return SchemaValidator(schema, config, _use_prebuilt=_use_prebuilt)
+
+
+class PluggableSchemaValidator:
+    """Pluggable schema validator."""
+
+    __slots__ = '_schema_validator', 'validate_json', 'validate_python', 'validate_strings'
+
+    def __init__(
+        self,
+        schema: CoreSchema,
+        schema_type: Any,
+        schema_type_path: SchemaTypePath,
+        schema_kind: SchemaKind,
+        config: CoreConfig | None,
+        plugins: Iterable[PydanticPluginProtocol],
+        plugin_settings: dict[str, Any],
+        _use_prebuilt: bool = True,
+    ) -> None:
+        self._schema_validator = SchemaValidator(schema, config, _use_prebuilt=_use_prebuilt)
+
+        python_event_handlers: list[BaseValidateHandlerProtocol] = []
+        json_event_handlers: list[BaseValidateHandlerProtocol] = []
+        strings_event_handlers: list[BaseValidateHandlerProtocol] = []
+        for plugin in plugins:
+            try:
+                p, j, s = plugin.new_schema_validator(
+                    schema, schema_type, schema_type_path, schema_kind, config, plugin_settings
+                )
+            except TypeError as e:  # pragma: no cover
+                raise TypeError(f'Error using plugin `{plugin.__module__}:{plugin.__class__.__name__}`: {e}') from e
+            if p is not None:
+                python_event_handlers.append(p)
+            if j is not None:
+                json_event_handlers.append(j)
+            if s is not None:
+                strings_event_handlers.append(s)
+
+        self.validate_python = build_wrapper(self._schema_validator.validate_python, python_event_handlers)
+        self.validate_json = build_wrapper(self._schema_validator.validate_json, json_event_handlers)
+        self.validate_strings = build_wrapper(self._schema_validator.validate_strings, strings_event_handlers)
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._schema_validator, name)
+
+
+def build_wrapper(func: Callable[P, R], event_handlers: list[BaseValidateHandlerProtocol]) -> Callable[P, R]:
+    if not event_handlers:
+        return func
+    else:
+        on_enters = tuple(h.on_enter for h in event_handlers if filter_handlers(h, 'on_enter'))
+        on_successes = tuple(h.on_success for h in event_handlers if filter_handlers(h, 'on_success'))
+        on_errors = tuple(h.on_error for h in event_handlers if filter_handlers(h, 'on_error'))
+        on_exceptions = tuple(h.on_exception for h in event_handlers if filter_handlers(h, 'on_exception'))
+
+        @functools.wraps(func)
+        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
+            for on_enter_handler in on_enters:
+                on_enter_handler(*args, **kwargs)
+
+            try:
+                result = func(*args, **kwargs)
+            except ValidationError as error:
+                for on_error_handler in on_errors:
+                    on_error_handler(error)
+                raise
+            except Exception as exception:
+                for on_exception_handler in on_exceptions:
+                    on_exception_handler(exception)
+                raise
+            else:
+                for on_success_handler in on_successes:
+                    on_success_handler(result)
+                return result
+
+        return wrapper
+
+
+def filter_handlers(handler_cls: BaseValidateHandlerProtocol, method_name: str) -> bool:
+    """Filter out handler methods which are not implemented by the plugin directly - e.g. those that are missing
+    or are inherited from the protocol.
+    """
+    handler = getattr(handler_cls, method_name, None)
+    if handler is None:
+        return False
+    elif handler.__module__ == 'pydantic.plugin':
+        # this is the original handler, from the protocol due to runtime inheritance
+        # we don't want to call it
+        return False
+    else:
+        return True