flask：如何对请求 JSON 和 JSON 模式进行验证？

Question

在 flask-restplus API 中，我需要对请求 JSON 数据进行验证，其中我已经使用 api.model 定义了请求正文模式。基本上我想将输入 JSON 数据传递给 API 函数，在使用 API 函数之前我必须验证输入 JSON 数据。为此，我使用RequestParser 来执行此任务，但是在验证和解析请求 JSON 之后，API 函数期望正确的 JSON 数据作为参数。要进行请求 JSON 验证，首先我必须解析接收到的输入 JSON 数据，解析其 JSON 主体，验证每一个，然后将其重构为 JSON 对象，然后传递给 API 函数。有没有更简单的方法来做到这一点？

输入 JSON 数据

{
  "body": {
    "gender": "M",
    "PCT": {
      "value": 12,
      "device": "roche_cobas"
    },
    "IL6": {
      "value": 12,
      "device": "roche_cobas"
    },
    "CRP": {
      "value": 12,
      "device": "roche_cobas"
    }
  }
}

我目前在烧瓶中的尝试

from flask_restplus import Api, Namespace, Resource, fields, reqparse, inputs
from flask import Flask, request, make_response, Response, jsonify

app = Flask(__name__)
api = Api(app)
ns = Namespace('')

feature = api.model('feature', {
    'value': fields.Integer(required=True),
    'time': fields.Date(required=True)
})

features = api.model('featureList', {
    'age': fields.String,
    'gender': fields.String(required=True),
    'time': fields.Date,
    'features': fields.List(fields.Nested(feature, required=True))
    })

@ns.route('/hello')
class helloResource(Resource):
    @ns.expect(features)
    def post(self):
        json_dict = request.json  ## get input JSON data
        ## need to parse json_dict to validate expected argument in JSON body
        root_parser = reqparse.RequestParser()
        root_parser.add_argument('body', type=dict)
        root_args = root_parser.parse_args()

        jsbody_parser = reqparse.RequestParser()
        jsbody_parser.add_argument('age', type=dict, location = ('body',))
        jsbody_parser.add_argument('gender', type=dict, location=('body',))
        ## IL6, CRP could be something else, how to use **kwargs here
        jsbody_parser.add_argument('IL6', type=dict, location=('body',))
        jsbody_parser.add_argument('PCT', type=dict, location=('body',))
        jsbody_parser.add_argument('CRP', type=dict, location=('body',))
        jsbody_parser = jsbody_parser.parse_args(req=root_args)

        ## after validate each argument on input JSON request body, all needs to be constructed as JSON data
        json_data = json.dumps(jsonify(jsbody_parser))   ## how can I get JSON again from jsbody_parser
        func_output = my_funcs(json_data)
        rest = make_response(jsonify(str(func_output)), 200)
        return rest

if __name__ == '__main__':
    api.add_namespace(ns) 
    app.run(debug=True)

更新：虚拟 api 函数

这是验证后需要 json 数据的虚拟函数：

import json
def my_funcs(json_data):
    a =json.loads(json_data)
    for k,v in a.iteritems():
           print k,v
    return jsonify(a)

上述尝试的当前输出：

我在响应正文中有这个：

{
  "errors": {
    "gender": "'gender' is a required property"
  },
  "message": "Input payload validation failed"
}

显然，请求 JSON 输入在我的尝试中没有得到处理和验证。我想我必须将 json_dict 传递给 RequestParser 对象，但仍然无法在此处验证请求 JSON。如何做到这一点？

我必须验证来自 JSON 主体的预期参数，在验证之后，我想构建将用作 API 函数参数的 JSON 主体。我怎样才能做到这一点？任何解决方法来实现这一点？

解析后的 JSON 必须传递给 my_funcs

在我的帖子中，请求 JSON 数据应该被解析，例如 age，gender 应该被验证为请求 JSON 中的预期参数，然后 jsonify 将添加的参数作为 JSON 并传递 my_funcs。如何在 fl 中轻松实现这一点

我想确保烧瓶应该解析 JSON 正文并按预期添加参数，否则会抛出错误。例如：

{
      "body": {
        "car": "M",
        "PCT": {
          "value": 12,
          "device": "roche_cobas"
        },
        "IL6": {
          "device": "roche_cobas"
        },
        "CRP": {
          "value": 12
        }
      }
    }

如果我提供像上面这样的 JSON 数据来向服务器端点发出 POST 请求，它应该会给出错误。如何做到这一点？如何验证烧瓶 API 调用的 POST 请求 JSON？

Answer 1

正如我试图在我们的谈话中表达的那样，您似乎正在使用序列化和反序列化工具。我发现 Marshmallow 是一个特殊的工具（它不是唯一的）。这是一个使用 Marshmallow 验证请求正文、将验证数据转换回 JSON 字符串并将其传递给函数进行操作并返回包含 JSON 数据的响应的工作示例：

from json import dumps, loads
from flask import Flask, jsonify, request
from marshmallow import Schema, fields, ValidationError

class BaseSchema(Schema):
    age = fields.Integer(required=True)
    gender = fields.String(required=True)

class ExtendedSchema(BaseSchema):
    # have a look at the examples in the Marshmallow docs for more complex data structures, such as nested fields.
    IL6 = fields.String()
    PCT = fields.String()
    CRP = fields.String()

def my_func(json_str:str):
    """ Your Function that Requires JSON string"""

    a_dict = loads(json_str)

    return a_dict

app = Flask(__name__)

@app.route('/base', methods=["POST"])
def base():
    # Get Request body from JSON
    request_data = request.json
    schema = BaseSchema()
    try:
        # Validate request body against schema data types
        result = schema.load(request_data)
    except ValidationError as err:
        # Return a nice message if validation fails
        return jsonify(err.messages), 400

    # Convert request body back to JSON str
    data_now_json_str = dumps(result)

    response_data = my_func(data_now_json_str)

    # Send data back as JSON
    return jsonify(response_data), 200


@app.route('/extended', methods=["POST"])
def extended():
    """ Same as base function but with more fields in Schema """
    request_data = request.json
    schema = ExtendedSchema()
    try:
        result = schema.load(request_data)
    except ValidationError as err:
        return jsonify(err.messages), 400

    data_now_json_str = dumps(result)

    response_data = my_func(data_now_json_str)

    return jsonify(response_data), 200

这里有一些显示验证的快速测试，以及扩展请求正文中的字段：

import requests
# Request fails validation
base_data = {
    'age': 42,
    }
r1 = requests.post('http://127.0.0.1:5000/base', json=base_data)
print(r1.content)

# Request passes validation
base_data = {
    'age': 42,
    'gender': 'hobbit'
    }
r2 = requests.post('http://127.0.0.1:5000/base', json=base_data)
print(r2.content)

# Request with extended properties
extended_data = {
    'age': 42,
    'gender': 'hobbit',
    'IL6': 'Five',
    'PCT': 'Four',
    'CRP': 'Three'}

r3 = requests.post('http://127.0.0.1:5000/extended', json=extended_data)
print(r3.content)

希望这可以帮助您到达目的地。