使用 flask-restful RequestParser 进行嵌套验证

Question

使用flask-restful 微框架，我无法构建将验证嵌套资源的RequestParser。假设表单的预期 JSON 资源格式：

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

a_list 中的每一项都对应一个对象：

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

... 然后使用类似以下的表单创建一个 RequestParser：

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

...但是您将如何验证a_list 中每个字典的嵌套MyObjects？或者，这是错误的方法吗？

this 对应的 API 将每个 MyObject 本质上视为一个对象字面量，并且可能有一个或多个传递给服务；因此，扁平化资源格式不适用于这种情况。

Answer 1

我建议使用cerberus 等数据验证工具。您首先为您的对象定义验证模式（嵌套对象模式在this 段落中介绍），然后使用验证器根据模式验证资源。当验证失败时，您还会收到详细的错误消息。

在以下示例中，我想验证位置列表：

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))

参数定义如下：

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

Answer 2

评分最高的解决方案不支持'strict=True'，要解决'strict=True'不支持的问题，可以创建一个FakeRequest对象来欺骗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)

顺便说一句：flask restful 会撕掉 RequestParser，并用 Marshmallow 替换它 Linkage

Answer 3

我通过为嵌套对象创建RequestParser 实例取得了成功。像往常一样首先解析根对象，然后使用结果输入嵌套对象的解析器。

诀窍是add_argument 方法的location 参数和parse_args 方法的req 参数。它们让您可以操纵RequestParser 所查看的内容。

这是一个例子：

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 4

我发现bbenne10s answer 真的很有用，但它对我来说并不适用。

我这样做的方式可能是错误的，但它确实有效。我的问题是我不明白action='append' 做了什么，因为它似乎做的是 wrap 列表中收到的值，但这对我来说没有任何意义。有人可以解释一下这在 cmets 中有什么意义吗？

所以我最终做的是创建自己的listtype，在value 参数中获取列表，然后以这种方式遍历列表：

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)

不是一个真正优雅的解决方案，但至少它可以工作。我希望有人能指出我们正确的方向......

更新

正如bbenne10 has said in the comments，action='append' 所做的是将所有同名的参数附加到一个列表中，因此对于 OP，它似乎不是很有用。

我已经迭代了我的解决方案，因为我不喜欢 reqparse 没有解析/验证任何嵌套对象的事实，所以我所做的是在自定义对象类型 @987654330 中使用 reqparse @。

首先，我声明了Request 的新子类，在解析嵌套对象时将其作为请求传递：

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

这个类覆盖了request.json，所以它使用一个新的json来解析对象。 然后，我在myobjlist 中添加了一个reqparse 解析器来解析所有参数，并添加了一个except 来捕获解析错误并传递reqparse 消息。

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

这样，即使是嵌套对象也会通过 reqparse 被解析并显示其错误

Answer 5

由于这里的 type 参数只不过是一个可调用的，它要么返回解析的值，要么在无效类型上引发 ValueError，我建议为此创建自己的类型验证器。验证器可能类似于：

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')