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

Posted 2023-02-23

【中文标题】使用 flask-restful RequestParser 进行嵌套验证

【英文标题】：Nested validation with the flask-restful RequestParser

【发布时间】：2013-10-14 14:39:27

【问题描述】：

使用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 本质上视为一个对象字面量，并且可能有一个或多个传递给服务；因此，扁平化资源格式不适用于这种情况。

【问题讨论】：

你解决了吗？如果是这样，请你提供一个自我回答，我在这里遇到同样的问题......提前谢谢。

不，抱歉。在研究了所有选项后，我选择了 Django REST Framework。

【参考方案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')

【讨论】：

这是一个非常好的解决方案。上面所有世界的最佳组合，感谢 kardaj。

【参考方案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

【参考方案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)

【讨论】：

谢谢@barqshasbite - 你能举个例子说明你的数据结构是什么样的吗？

@barqshasbite 解释得很好。

这个变通方法不支持'strict=True'，如果你用'strict=True'解析参数，它就会失败。

【参考方案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 被解析并显示其错误

【讨论】：

action="append" 说明使用多次指定的键指定 HTTP 请求是完全有效的（即允许对 http://example.com/api?x=1&x=2 的 GET 请求）。 Flask 通过为每个键使用 MultiDict 来做到这一点，action="append" 只是将此行为公开给 Web 开发人员。我想你想要的是接受一个看起来像MyObj 的东西的列表。如果是这样的话，我认为你的解决方案是干净地解决问题的唯一真正方法。

（此外，我相信我只是在上面的代码中真正指定了action="append"，因为它存在于 Daniel 提供的示例代码中：P）

谢谢。现在它是有道理的。所以我理解这个问题的方式，我的答案应该有效:)然后我不会再担心我的实现是否足够好，因为它看起来如此：D

谢谢。救了我的培根。

【参考方案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')