Nested validation with the flask-restful RequestParser

Question

Using the flask-restful micro-framework, I am having trouble constructing a RequestParser that will validate nested resources. Assuming an expected JSON resource format of the form:

{
    'a_list': [
        {
            'obj1': 1,
            'obj2': 2,
            'obj3': 3
        },
        {
            'obj1': 1,
            'obj2': 2,
            'obj3': 3
        }
    ]
}

Each item in a_list corresponds to an object:

class MyObject(object):
    def __init__(self, obj1, obj2, obj3)
        self.obj1 = obj1
        self.obj2 = obj2
        self.obj3 = obj3

... and one would then create a RequestParser using a form something like:

from flask.ext.restful import reqparse
parser = reqparse.RequestParser()
parser.add_argument('a_list', type=MyObject, action='append')

... but how would you validate the nested MyObjects of each dictionary inside a_list? Or, alternately, is this the wrong approach?

The API this corresponds to treats each MyObject as, essentially, an object literal, and there may be one or more of them passed to the service; therefore, flattening the resource format will not work for this circumstance.

Answer 1

I used pydantic as it's very pythonic, especially in Python >= 3.10:

Create Validation models:

from pydantic import BaseModel

class SecondLevel(BaseModel):
    obj1: int
    obj2: int
    obj3: int

class FirstLevel(BaseModel):
    a_list: list[SecondLevel]

Create validation function for the parser:

from json import JSONDecodeError
from pydantic import ValidationError

def my_validator(value):
    try:
        data = json.loads(value)  # convert to object
        obj = FirstLevel.parse_obj(data)  # actually parse data
        return obj
    except JSONDecodeError:
        raise ValueError(f"Could not parse JSON string.")
    except ValidationError as e:
        print(e)
        raise ValueError(f"Could not validate the object json. {e}")

Then, add the validator to the parser:

parser.add_argument("a_list", type=my_validator)

It works like intended, supports default values, missing values and says what's missing in the thrown exception.

Answer 2

After spending some time trying to understand the parse_args function, and considering the example of the question, a solution for validating nested arguments might be:

list_args_parser = reqparse.RequestParser()
list_args_parser.add_argument('a_list', type=dict, action='append', default=[])

list_item_args_parser = reqparse.RequestParser()
list_items_args_parser.add_argument('obj1', type=int, location='json')
list_items_args_parser.add_argument('obj2', type=int, location='json')
list_items_args_parser.add_argument('obj3', type=int, location='json')

then do:

from types import SimpleNamespace

parsed_list_items = list()
list_items = list_args_parser.parse_args().pop('a_list')
for i in list_items:
    parsed_list_item.append(list_item_args_parser.parse_args(SimpleNamespace(**{'json': i})))

      

or in a one-line version:

a_list_items = list_args_parse.parse_args()
parsed_list_items = list(map(lambda i: list_item_args_parse.parse_args(SimpleNamespace(**{'json': i})), a_list_items.pop('a_list'))

The trick is to call parse_args passing SimpleNamespace(**{'json': <list_item>}) as argument for each item in the list. Don't forget to set location='json' for each argument in the list as shown in the example

Hope this helps

Answer 3

import jsonschema
from jsonschema import validate

def validate_json_request(jsonData, schema):
    try:
        validate(instance=jsonData, schema=schema)
    except jsonschema.exceptions.ValidationError as err:
        return False
    return True


@api.resource('/product-catalog-api/accounts/<string:domain>/selected/list', endpoint='product-list-varient-resolver-apis')
class ProductVarientRealAPI(Resource):
    def __init__(self) -> None:
        '''
            {
                "products": [
                    {
                        "baseCode": "cEMbseQRA-fJGo-YD2ggc_base00",
                        "selectedVarientList": [
                            "Hoe60Ypbxzxd0aMRmV4Ff",
                            "gMVWwbxfPSyaXQjRLX4Sv"
                        ]
                    },
                    {
                        "baseCode": "cEMbseQRA-fJGo-YD2ggc_base00",
                        "selectedVarientList": [
                            "UzsdFS7ZgFPTUnwswVpuq",
                            "gMVWwbxfPSyaXQjRLX4Sv"
                        ]
                    }
                ]
            }
        '''
        
        self.reqparse = reqparse.RequestParser()
        self.reqparse.add_argument(
            'products', required=True, type=dict, help="need [products]", action="append")

    def post(self, domain):
        products_schema = {
            'type': 'array',
            'items': {
                'type': 'object',
                'properties': {
                    'baseCode': {
                        'type': 'string',
                    },
                    'selectedVarientList': {
                        'type': 'array',
                        'items': {'type':'string'}
                    }
                }
            }
        }
        
        try:
            args = self.reqparse.parse_args()
            if not validate_json_request(args.products, products_schema): # check like this
                raise Exception("Invalid Data")
            product_list = []
            for product in args.products:
                product_info, status = get_product_varient(product["baseCode"], domain, product["selectedVarientList"])
                if status == 200:
                    product_list.append(product_info)
            return {
                'api_response_info': {
                    'message': 'Fetched Product Varients',
                },
                'data': product_list
            }, 200
        except Exception as err:
            print("error occ", err)
            return {
                'message': 'Error: {}'.format(err)
            }, 500

i tried above methods, in the posts this is more readable and better i think, it gives you a better abstraction layer. but here is 2 type of validations going which might make this req slow. one is regparser which is slow, and jsonschema is also slow**

Answer 4

I would suggest using a data validation tool such as cerberus. You start by defining a validation schema for your object (Nested object schema is covered in this paragraph), then use a validator to validate the resource against the schema. You also get detailed error messages when the validation fails.

In the following example, I want to validate a list of locations:

from cerberus import Validator
import json


def location_validator(value):
    LOCATION_SCHEMA = {
        'lat': {'required': True, 'type': 'float'},
        'lng': {'required': True, 'type': 'float'}
    }
    v = Validator(LOCATION_SCHEMA)
    if v.validate(value):
        return value
    else:
        raise ValueError(json.dumps(v.errors))

The argument is defined as follows:

parser.add_argument('location', type=location_validator, action='append')

Answer 5

The highest rated solution do not support 'strict=True', To solve the 'strict=True' not support issue, you can create a FakeRequest object to cheat RequestParser

class FakeRequest(dict):
    def __setattr__(self, name, value):
        object.__setattr__(self, name, value)

root_parser = reqparse.RequestParser()
root_parser.add_argument('id', type=int)
root_parser.add_argument('name', type=str)
root_parser.add_argument('nested_one', type=dict)
root_parser.add_argument('nested_two', type=dict)
root_args = root_parser.parse_args()

nested_one_parser = reqparse.RequestParser()
nested_one_parser.add_argument('id', type=int, location=('json',))

fake_request = FakeRequest()
setattr(fake_request, 'json', root_args['nested_one'])
setattr(fake_request, 'unparsed_arguments', {})

nested_one_args = nested_one_parser.parse_args(req=fake_request, strict=True)

fake_request = FakeRequest()
setattr(fake_request, 'json', root_args['nested_two'])
setattr(fake_request, 'unparsed_arguments', {})

nested_two_parser = reqparse.RequestParser()
nested_two_parser.add_argument('id', type=int, location=('json',))
nested_two_args = nested_two_parser.parse_args(req=fake_request, strict=True)

BTW: flask restful will rip RequestParser out, and replace it with Marshmallow Linkage

Answer 6

I have had success by creating RequestParser instances for the nested objects. Parse the root object first as you normally would, then use the results to feed into the parsers for the nested objects.

The trick is the location argument of the add_argument method and the req argument of the parse_args method. They let you manipulate what the RequestParser looks at.

Here's an example:

root_parser = reqparse.RequestParser()
root_parser.add_argument('id', type=int)
root_parser.add_argument('name', type=str)
root_parser.add_argument('nested_one', type=dict)
root_parser.add_argument('nested_two', type=dict)
root_args = root_parser.parse_args()

nested_one_parser = reqparse.RequestParser()
nested_one_parser.add_argument('id', type=int, location=('nested_one',))
nested_one_args = nested_one_parser.parse_args(req=root_args)

nested_two_parser = reqparse.RequestParser()
nested_two_parser.add_argument('id', type=int, location=('nested_two',))
nested_two_args = nested_two_parser.parse_args(req=root_args)

Answer 7

I found the bbenne10s answer really useful, but it didn't work for me as is.

The way I did this is probably wrong, but it works. My problem is that I don't understand what action='append' does as what it seems to do is wrap the value received in a list, but it doesn't make any sense to me. Can someone please explain whats the point of this in the comments?

So what I ended up doing is creating my own listtype, get the list inside the value param and then iterate through the list this way:

from flask.ext.restful import reqparse
def myobjlist(value):
    result = []
    try:
        for v in value:
            x = MyObj(**v)
            result.append(x)
    except TypeError:
        raise ValueError("Invalid object")
    except:
        raise ValueError

    return result


#and now inside views...
parser = reqparse.RequestParser()
parser.add_argument('a_list', type=myobjlist)

Not a really elegant solution, but at least it does the work. I hope some one can point us in the right direction...

Update

As bbenne10 has said in the comments, what action='append' does is append all the arguments named the same into a list, so in the case of the OP, it doesn't seem to be very useful.

I have iterated over my solution because I didn't like the fact that reqparse wasn't parsing/validating any of the nested objects so I what I have done is use reqparse inside the custom object type myobjlist.

First, I have declared a new subclass of Request, to pass it as the request when parsing the nested objects:

class NestedRequest(Request):
    def __init__(self, json=None, req=request):
        super(NestedRequest, self).__init__(req.environ, False, req.shallow)
        self.nested_json = json

    @property
    def json(self):
        return self.nested_json

This class overrides the request.json so that it uses a new json with the object to being parsed. Then, I added a reqparse parser to myobjlist to parse all the arguments and added an except to catch the parsing error and pass the reqparse message.

from flask.ext.restful import reqparse
from werkzeug.exceptions import ClientDisconnected
def myobjlist(value):
    parser = reqparse.RequestParser()
    parser.add_argument('obj1', type=int, required=True, help='No obj1 provided', location='json')
    parser.add_argument('obj2', type=int, location='json')
    parser.add_argument('obj3', type=int, location='json')
    nested_request = NestedRequest()
    result = []
    try:
        for v in value:
            nested_request.nested_json = v
            v = parser.parse_args(nested_request)
            x = MyObj(**v)
            result.append(x)
    except TypeError:
        raise ValueError("Invalid object")
    except ClientDisconnected, e:
        raise ValueError(e.data.get('message', "Parsing error") if e.data else "Parsing error")
    except:
        raise ValueError
    return result

This way, even the nested objects will get parsed through reqparse and will show its errors

Answer 8

Since the type argument here is nothing but a callable that either returns a parsed value or raise ValueError on invalid type, I would suggest creating your own type validator for this. The validator could look something like:

from flask.ext.restful import reqparse
def myobj(value):
    try:
        x = MyObj(**value)
    except TypeError:
        # Raise a ValueError, and maybe give it a good error string
        raise ValueError("Invalid object")
    except:
        # Just in case you get more errors
        raise ValueError 

    return x


#and now inside your views...
parser = reqparse.RequestParser()
parser.add_argument('a_list', type=myobj, action='append')