33

フラスコ レストフルマイクロ フレームワークを使用して、ネストされたリソースを検証する を構築するのに問題があります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...しかし、内部の各辞書のネストされた をどのように検証しますa_listか? あるいは、これは間違ったアプローチですか?

これが対応する API は、それぞれMyObjectを本質的にオブジェクト リテラルとして扱い、それらの 1 つ以上がサービスに渡される場合があります。したがって、リソース形式をフラット化しても、この状況では機能しません。

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 に答える
4

bbenne10sの回答は非常に役に立ちましたが、そのままでは機能しませんでした。

私がこれを行った方法はおそらく間違っていますが、うまくいきます。私の問題は、受け取った値をリストにラップaction='append'しているように見えるので、何が何をするのか理解できないことですが、私には意味がありません。誰かがコメントでこれのポイントを説明できますか?

だから私がやったのは、自分の を作成し、 paramlisttype内のリストを取得してから、次のようにリストを反復処理することです: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 に答える
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 種類の検証が行われているため、この要件が遅くなる可能性があります。1つは遅いregparserであり、jsonschemaも遅い**

于 2022-02-06T18:11:40.100 に答える