33

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

...但是您将如何验证MyObject每个字典的嵌套 s 内部a_list?或者,或者,这是错误的方法吗?

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

4

6 回答 6

31

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

诀窍是方法的参数和方法location的参数。他们让你操纵外观。add_argumentreqparse_argsRequestParser

这是一个例子:

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)
于 2014-11-23T17:26:56.120 回答
9

我建议使用数据验证工具,例如cerberus。您首先为您的对象定义一个验证模式(段介绍了嵌套对象模式),然后使用验证器根据该模式验证资源。当验证失败时,您还会收到详细的错误消息。

在以下示例中,我想验证位置列表:

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')
于 2016-06-15T20:29:49.280 回答
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')
于 2014-01-17T18:41:00.100 回答
4

我发现bbenne10s 的答案真的很有用,但它对我不起作用。

我这样做的方式可能是错误的,但它确实有效。我的问题是我不明白action='append'它似乎做了什么是将收到的值包装在一个列表中,但这对我来说没有任何意义。有人可以在评论中解释这是什么意思吗?

所以我最终做的是创建我自己的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 在评论中所说action='append'所做的是将所有名称相同的参数附加到一个列表中,所以在 OP 的情况下,它似乎不是很有用。

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

首先,我声明了一个新的 , 子类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 和要解析的对象。然后,我添加了一个reqparse解析器来myobjlist解析所有参数,并添加了一个异常来捕获解析错误并传递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 进行解析并显示其错误

于 2014-07-04T12:12:04.453 回答
3

评分最高的解决方案不支持'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替换它

于 2016-05-12T09:50:05.913 回答
0
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

我尝试了上述方法,在帖子中我认为这更具可读性和更好,它为您提供了更好的抽象层。但这里有 2 种类型的验证可能会使这个请求变慢。一个是regparser,它很慢,jsonschema也很慢**

于 2022-02-06T18:11:40.100 回答