ryg/royalnet
ryg/royalnet
1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-22 19:14:20 +00:00
Code Activity

Refactor royalnet.engineer.teleporter into royalnet.validation

Browse source
This commit is contained in:
Steffo 2022-05-02 16:46:09 +02:00
parent e717d1d0cb
commit 8fed7ba510
Signed by: steffo
GPG key ID: 6965406171929D01
3 changed files with 320 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

66
poetry.lock generated
View file

@ -173,6 +173,21 @@ category = "dev"
optional = false
python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*"
[[package]]
name = "pydantic"
version = "1.9.0"
description = "Data validation and settings management using python 3.6 type hinting"
category = "main"
optional = false
python-versions = ">=3.6.1"
[package.dependencies]
typing-extensions = ">=3.7.4.3"
[package.extras]
dotenv = ["python-dotenv (>=0.10.4)"]
email = ["email-validator (>=1.0.3)"]
[[package]]
name = "pygments"
version = "2.12.0"
@ -406,6 +421,14 @@ category = "dev"
optional = false
python-versions = ">=3.7"
[[package]]
name = "typing-extensions"
version = "4.2.0"
description = "Backported and Experimental Type Hints for Python 3.7+"
category = "main"
optional = false
python-versions = ">=3.7"
[[package]]
name = "urllib3"
version = "1.26.9"
@ -422,7 +445,7 @@ socks = ["PySocks (>=1.5.6,!=1.5.7,<2.0)"]
[metadata]
lock-version = "1.1"
python-versions = "^3.10"
content-hash = "f1b32330c1cee63fc4428facd3a415272d4123a2a8d0222c0847c6acc4052dca"
content-hash = "9404da8b8b5119e177b72064f2e83cc13a6177a27fdb7acd896d682b16bbc048"
[metadata.files]
alabaster = [
@ -574,6 +597,43 @@ py = [
{file = "py-1.11.0-py2.py3-none-any.whl", hash = "sha256:607c53218732647dff4acdfcd50cb62615cedf612e72d1724fb1a0cc6405b378"},
{file = "py-1.11.0.tar.gz", hash = "sha256:51c75c4126074b472f746a24399ad32f6053d1b34b68d2fa41e558e6f4a98719"},
]
pydantic = [
{file = "pydantic-1.9.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:cb23bcc093697cdea2708baae4f9ba0e972960a835af22560f6ae4e7e47d33f5"},
{file = "pydantic-1.9.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:1d5278bd9f0eee04a44c712982343103bba63507480bfd2fc2790fa70cd64cf4"},
{file = "pydantic-1.9.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ab624700dc145aa809e6f3ec93fb8e7d0f99d9023b713f6a953637429b437d37"},
{file = "pydantic-1.9.0-cp310-cp310-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:c8d7da6f1c1049eefb718d43d99ad73100c958a5367d30b9321b092771e96c25"},
{file = "pydantic-1.9.0-cp310-cp310-musllinux_1_1_i686.whl", hash = "sha256:3c3b035103bd4e2e4a28da9da7ef2fa47b00ee4a9cf4f1a735214c1bcd05e0f6"},
{file = "pydantic-1.9.0-cp310-cp310-musllinux_1_1_x86_64.whl", hash = "sha256:3011b975c973819883842c5ab925a4e4298dffccf7782c55ec3580ed17dc464c"},
{file = "pydantic-1.9.0-cp310-cp310-win_amd64.whl", hash = "sha256:086254884d10d3ba16da0588604ffdc5aab3f7f09557b998373e885c690dd398"},
{file = "pydantic-1.9.0-cp36-cp36m-macosx_10_9_x86_64.whl", hash = "sha256:0fe476769acaa7fcddd17cadd172b156b53546ec3614a4d880e5d29ea5fbce65"},
{file = "pydantic-1.9.0-cp36-cp36m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:c8e9dcf1ac499679aceedac7e7ca6d8641f0193c591a2d090282aaf8e9445a46"},
{file = "pydantic-1.9.0-cp36-cp36m-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:d1e4c28f30e767fd07f2ddc6f74f41f034d1dd6bc526cd59e63a82fe8bb9ef4c"},
{file = "pydantic-1.9.0-cp36-cp36m-musllinux_1_1_i686.whl", hash = "sha256:c86229333cabaaa8c51cf971496f10318c4734cf7b641f08af0a6fbf17ca3054"},
{file = "pydantic-1.9.0-cp36-cp36m-musllinux_1_1_x86_64.whl", hash = "sha256:c0727bda6e38144d464daec31dff936a82917f431d9c39c39c60a26567eae3ed"},
{file = "pydantic-1.9.0-cp36-cp36m-win_amd64.whl", hash = "sha256:dee5ef83a76ac31ab0c78c10bd7d5437bfdb6358c95b91f1ba7ff7b76f9996a1"},
{file = "pydantic-1.9.0-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:d9c9bdb3af48e242838f9f6e6127de9be7063aad17b32215ccc36a09c5cf1070"},
{file = "pydantic-1.9.0-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2ee7e3209db1e468341ef41fe263eb655f67f5c5a76c924044314e139a1103a2"},
{file = "pydantic-1.9.0-cp37-cp37m-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:0b6037175234850ffd094ca77bf60fb54b08b5b22bc85865331dd3bda7a02fa1"},
{file = "pydantic-1.9.0-cp37-cp37m-musllinux_1_1_i686.whl", hash = "sha256:b2571db88c636d862b35090ccf92bf24004393f85c8870a37f42d9f23d13e032"},
{file = "pydantic-1.9.0-cp37-cp37m-musllinux_1_1_x86_64.whl", hash = "sha256:8b5ac0f1c83d31b324e57a273da59197c83d1bb18171e512908fe5dc7278a1d6"},
{file = "pydantic-1.9.0-cp37-cp37m-win_amd64.whl", hash = "sha256:bbbc94d0c94dd80b3340fc4f04fd4d701f4b038ebad72c39693c794fd3bc2d9d"},
{file = "pydantic-1.9.0-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:e0896200b6a40197405af18828da49f067c2fa1f821491bc8f5bde241ef3f7d7"},
{file = "pydantic-1.9.0-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:7bdfdadb5994b44bd5579cfa7c9b0e1b0e540c952d56f627eb227851cda9db77"},
{file = "pydantic-1.9.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:574936363cd4b9eed8acdd6b80d0143162f2eb654d96cb3a8ee91d3e64bf4cf9"},
{file = "pydantic-1.9.0-cp38-cp38-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:c556695b699f648c58373b542534308922c46a1cda06ea47bc9ca45ef5b39ae6"},
{file = "pydantic-1.9.0-cp38-cp38-musllinux_1_1_i686.whl", hash = "sha256:f947352c3434e8b937e3aa8f96f47bdfe6d92779e44bb3f41e4c213ba6a32145"},
{file = "pydantic-1.9.0-cp38-cp38-musllinux_1_1_x86_64.whl", hash = "sha256:5e48ef4a8b8c066c4a31409d91d7ca372a774d0212da2787c0d32f8045b1e034"},
{file = "pydantic-1.9.0-cp38-cp38-win_amd64.whl", hash = "sha256:96f240bce182ca7fe045c76bcebfa0b0534a1bf402ed05914a6f1dadff91877f"},
{file = "pydantic-1.9.0-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:815ddebb2792efd4bba5488bc8fde09c29e8ca3227d27cf1c6990fc830fd292b"},
{file = "pydantic-1.9.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:6c5b77947b9e85a54848343928b597b4f74fc364b70926b3c4441ff52620640c"},
{file = "pydantic-1.9.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:4c68c3bc88dbda2a6805e9a142ce84782d3930f8fdd9655430d8576315ad97ce"},
{file = "pydantic-1.9.0-cp39-cp39-manylinux_2_5_i686.manylinux1_i686.manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:5a79330f8571faf71bf93667d3ee054609816f10a259a109a0738dac983b23c3"},
{file = "pydantic-1.9.0-cp39-cp39-musllinux_1_1_i686.whl", hash = "sha256:f5a64b64ddf4c99fe201ac2724daada8595ada0d102ab96d019c1555c2d6441d"},
{file = "pydantic-1.9.0-cp39-cp39-musllinux_1_1_x86_64.whl", hash = "sha256:a733965f1a2b4090a5238d40d983dcd78f3ecea221c7af1497b845a9709c1721"},
{file = "pydantic-1.9.0-cp39-cp39-win_amd64.whl", hash = "sha256:2cc6a4cb8a118ffec2ca5fcb47afbacb4f16d0ab8b7350ddea5e8ef7bcc53a16"},
{file = "pydantic-1.9.0-py3-none-any.whl", hash = "sha256:085ca1de245782e9b46cefcf99deecc67d418737a1fd3f6a4f511344b613a5b3"},
{file = "pydantic-1.9.0.tar.gz", hash = "sha256:742645059757a56ecd886faf4ed2441b9c0cd406079c2b4bee51bcc3fbcd510a"},
]
pygments = [
{file = "Pygments-2.12.0-py3-none-any.whl", hash = "sha256:dc9c10fb40944260f6ed4c688ece0cd2048414940f1cea51b8b226318411c519"},
{file = "Pygments-2.12.0.tar.gz", hash = "sha256:5eb116118f9612ff1ee89ac96437bb6b49e8f04d8a13b514ba26f620208e26eb"},
@ -646,6 +706,10 @@ tomli = [
{file = "tomli-2.0.1-py3-none-any.whl", hash = "sha256:939de3e7a6161af0c887ef91b7d41a53e7c5a1ca976325f429cb46ea9bc30ecc"},
{file = "tomli-2.0.1.tar.gz", hash = "sha256:de526c12914f0c550d15924c62d72abc48d6fe7364aa87328337a31007fe8a4f"},
]
typing-extensions = [
{file = "typing_extensions-4.2.0-py3-none-any.whl", hash = "sha256:6657594ee297170d19f67d55c05852a874e7eb634f4f753dbd667855e07c1708"},
{file = "typing_extensions-4.2.0.tar.gz", hash = "sha256:f1c24655a0da0d1b67f07e17a5e6b2a105894e6824b92096378bb3668ef02376"},
]
urllib3 = [
{file = "urllib3-1.26.9-py2.py3-none-any.whl", hash = "sha256:44ece4d53fb1706f667c9bd1c648f5469a2ec925fcf3a776667042d645472c14"},
{file = "urllib3-1.26.9.tar.gz", hash = "sha256:aabaf16477806a5e1dd19aa41f8c2b7950dd3c746362d7e3223dbe6de6ac448e"},

1
pyproject.toml
View file

@ -136,6 +136,7 @@ classifiers = [
python = "^3.10"
async-property = "^0.2.1"
pydantic = "^1.9.0"

254
royalnet/validation.py Normal file
View file

@ -0,0 +1,254 @@
"""
This module contains various objects used to perform validation on function inputs and outputs.
"""
import pydantic
import inspect
import logging
from .royaltyping import *
from .exc import RoyalnetException
log = logging.getLogger(__name__)
class ValidationError(RoyalnetException, pydantic.ValidationError):
"""
Base error for the :mod:`royalnet.validators` module.
"""
class InputValidationError(ValidationError):
"""
A failure in the validation of the function arguments.
"""
class OutputValidationError(ValidationError):
"""
A failure in the validation of the function return value.
"""
class ValidatingFunction:
"""
A function wrapper which uses :mod:`pydantic` to optionally perform type checking on arguments and return value.
"""
def __init__(self, f: Callable[..., Any], validate_input: bool = True, validate_output: bool = True):
self.f: Callable[..., Any] = f
"""
The function which is having its parameters and return value validated.
"""
self.InputModel: Type[pydantic.BaseModel] = self._create_input_model() if validate_input else None
"""
The :mod:`pydantic` model used to validate input parameters, or :data:`None` if they should not be validated.
"""
self.OutputModel: Type[pydantic.BaseModel] = self._create_output_model() if validate_output else None
"""
The :mod:`pydantic` model used to validate the return value, or :data:`None` if it shouldn't be validated.
"""
def __repr__(self):
if self.InputModel and self.OutputModel:
validation = "validating input and output"
elif self.InputModel:
validation = "validating only input"
elif self.OutputModel:
validation = "validating only output"
else:
validation = "not validating anything"
return f"<{self.__class__.__qualname__} {validation}>"
@staticmethod
def _parameter_to_field(param: inspect.Parameter, **kwargs) -> Tuple[type, pydantic.fields.FieldInfo]:
"""
Convert a :class:`inspect.Parameter` to a type-field :class:`tuple`, which can be easily passed to
:func:`pydantic.create_model`.
If the parameter is already a :class:`pydantic.FieldInfo` (created by :func:`pydantic.Field`), it will be
returned as the value, without creating a new model.
:param param: The :class:`inspect.Parameter` to convert.
:param kwargs: Additional kwargs to pass to the field.
:return: A :class:`tuple`, where the first element is a :class:`type` and the second is a :class:`pydantic.Field`.
"""
if isinstance(param.default, pydantic.fields.FieldInfo):
log.debug(f"Parameter {param} is a pydantic.Field, leaving it untouched...")
return (
param.annotation,
param.default
)
else:
log.debug(f"Parameter {param} is not a pydantic.Field, converting it to one...")
return (
param.annotation,
pydantic.Field(
default=param.default if param.default is not inspect.Parameter.empty else ...,
title=param.name,
**kwargs,
),
)
class ModelConfig(pydantic.BaseConfig):
"""
Configuration for the :mod:`pydantic` models generated by this class.
"""
arbitrary_types_allowed = True
def _create_input_model(self, **extra_fields) -> Type[pydantic.BaseModel]:
"""
Create a pydantic model based on the arguments of the :attr:`f` function.
Function arguments starting with ``_`` are ignored.
The model is created using the current :class:`.ModelConfig`.
:param extra_fields: Extra fields to be added to the model.
:return: The created pydantic model.
"""
log.debug(f"Getting function signature of: {self.f!r}")
signature: inspect.Signature = inspect.signature(self.f)
log.debug(f"Converting parameter annotations of {self.f!r} to fields...")
fields = {
key: self._parameter_to_field(value)
for key, value in signature.parameters.items()
if not key.startswith("_")
}
log.debug(f"Creating input model with parsed fields {fields!r} and extra fields {extra_fields!r}...")
return pydantic.create_model(
f"{self.__class__.__name__}InputModel",
__config__=self.ModelConfig,
**fields,
**extra_fields
)
def _create_output_model(self) -> Type[pydantic.BaseModel]:
"""
Create a pydantic model based on the return value of the :attr:`f` function.
The model is created using the current :class:`.ModelConfig`.
:return: The created pydantic model.
"""
log.debug(f"Getting function signature of: {self.f!r}")
signature: inspect.Signature = inspect.signature(self.f)
log.debug(f"Creating output model...")
return pydantic.create_model(
f"{self.__class__.__name__}OutputModel",
__config__=self.ModelConfig,
__root__=(signature.return_annotation, pydantic.Field(..., title="Returns"))
)
def _validate_input(self, **kwargs) -> pydantic.BaseModel:
"""
Instantiate the :attr:`.InputModel` with the passed kwargs.
:param kwargs: The keyword arguments that should be passed to the model.
:return: The created model.
:raises .InputValidationError: If the kwargs fail the validation.
"""
log.debug(f"Validating input: {kwargs!r}")
try:
return self.InputModel(**kwargs)
except pydantic.ValidationError as e:
log.error(f"Input validation failed: {e!r}")
raise InputValidationError(errors=e.raw_errors, model=e.model)
def teleport_out(self, value: Any) -> pydantic.BaseModel:
"""
Instantiate the :attr:`.OutputModel` with the passed value.
:param value: The value that should be validated.
:return: The created model.
:raises .OutputValidationError: If the value fails the validation.
"""
log.debug(f"Validating output: {value!r}")
try:
return self.OutputModel(__root__=value)
except pydantic.ValidationError as e:
log.error(f"Output validation failed: {e!r}")
raise OutputValidationError(errors=e.raw_errors, model=e.model)
@staticmethod
def _split_kwargs(**kwargs) -> Tuple[Dict[str, Any], Dict[str, Any]]:
"""
Split the passed kwargs in two different :class:`dict`:
- One containing the arguments that **do not start with ``_``**;
- Another containing the ones which do.
:return: A tuple of :class:`dict`, where the second contains the ones starting with ``_``, and the first
contains the rest.
"""
model_params = {}
extra_params = {}
for key, value in kwargs.items():
if key.startswith("_"):
log.debug(f"Found extra keyword argument: {key}")
extra_params[key] = value
else:
log.debug(f"Found model keyword argument: {key}")
model_params[key] = value
return model_params, extra_params
def _run(self, **kwargs) -> Any:
"""
Run the :class:`.ValidatingFunction` synchronously.
"""
if self.InputModel:
log.debug("Validating input...")
model_kwargs, extra_kwargs = self._split_kwargs(**kwargs)
model_kwargs = self._validate_input(**kwargs).dict()
kwargs = {**model_kwargs, **extra_kwargs}
result = self.f(**kwargs)
if self.OutputModel:
# noinspection PyUnresolvedReferences
result = self.teleport_out(result).__root__
return result
def _run_async(self, **kwargs) -> Awaitable[Any]:
"""
Run the :class:`.ValidatingFunction` asynchronously.
"""
if self.InputModel:
log.debug("Validating input...")
model_kwargs, extra_kwargs = self._split_kwargs(**kwargs)
model_kwargs = self._validate_input(**kwargs).dict()
kwargs = {**model_kwargs, **extra_kwargs}
result = await self.f(**kwargs)
if self.OutputModel:
# noinspection PyUnresolvedReferences
result = self.teleport_out(result).__root__
return result
def __call__(self, *args, **kwargs):
if inspect.iscoroutinefunction(self.f):
return self._run_async(**kwargs)
else:
return self._run(**kwargs)
__all__ = (
"ValidationError",
"InputValidationError",
"OutputValidationError",
"ValidatingFunction",
)