Running async procedure in custom validator

[rmedaer]

Hi,

I'm using jsonschema for a while... Going deeper in schema validation I figured out that I can define my own validation procedures.(Thanks Radek Lat for his nice article)

So one of the feature I'd like to use is the custom validators. However (in my use case), it would be useful only if I can run asynchronous task. Furthermore I'm running validation from a coroutine.

So here is my question: How can I execute async calls in a custom validator ? Does the generator can help me ?

[Julian]

Think I mentioned via PM that this is a long term goal (to better support async use cases, more so for ref resolution than validation but sure there too). Closing this for now but of course happy to review any proposed changes.

[rmedaer]

I think I finally found an elegant way to do mixed async/sync validation. Indeed it needs some changes but few ! Here is my proposal...

The idea would be to declare a new type of ValidationError to yield during the validation (for instance AsyncValidation) . This new "pseudo validation error" may go through the generator validation tree until a Validator#async_iter_errors method where we may check if it isinstance(error, AsyncValidation). If yes, we shall call asynchronously error.async_validate which will populate a buffer of errors later yielded by the original validator method. Pay attention that it will NOT work out of the box with "conditional checks" (like anyOf, ..) but I'm pretty sure that there is not a lot of changes to do to handle these cases.

If AsyncValidation is not supported (for instance if we "only" call iter_errors) it will generate a validation error.

Quiet confusing ? Ok, here is an example:

import asyncio
import functools
from jsonschema import ValidationError, Draft7Validator
from jsonschema.validators import extend


class AsyncValidation(ValidationError):

    def __init__(self, callback, value, *args, message='async validation not supported', **kwargs):
        super(AsyncValidation, self).__init__(message, *args, **kwargs)
        self.callback = callback
        self.value = value
        self.errors = []

    async def async_validate(self):
        async for error in self.callback(self.validator, self.value, self.instance, self.schema):
            self.errors.append(error)

class AsyncValidator():

    def __init__(self, draft_src, validators, schema):
        validator = extend(draft_src, validators)(schema)
        self.__class__ = type(validator.__class__.__name__, (self.__class__, validator.__class__), {})
        self.__dict__ = validator.__dict__

    async def async_iter_errors(self, body):
        for error in self.iter_errors(body):
            if isinstance(error, AsyncValidation):
                await error.async_validate()
            else:
                yield error

def wrap_async_validator(coroutine):
    def co_validator(coroutine, validator, value, instance, schema):
        async_validation = AsyncValidation(coroutine, value=value, validator=validator, instance=instance, schema=schema)
        yield async_validation
        for error in async_validation.errors:
            yield error

    return functools.partial(co_validator, coroutine)

async def custom_validation(validator, value, instance, schema):
    await asyncio.sleep(5)
    yield ValidationError('hmmm, something goes wrong here...')

async def async_validation_example():
    body = {'x': '...'}
    schema = {'type': 'object', 'properties': {'x': {'type': 'string', 'custom': True}}}
    validator = AsyncValidator(
        Draft7Validator,
        {'custom': wrap_async_validator(custom_validation)},
        schema,
    )

    async for error in validator.async_iter_errors(body):
        print(str(error))

I would enjoy to have your feedback about this idea ! ;-)

Kind regards,