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

Question

在 flask-restplus API 中，我需要对请求 JSON 数据进行验证，其中我已经用api.model. 基本上我想将输入的 JSON 数据传递给 API 函数，在使用 API 函数之前我必须验证输入的 JSON 数据。为此，我用于RequestParser执行此任务，但 API 函数期望在验证和解析请求 JSON 后将正确的 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. 如何在佛罗里达州轻松做到这一点

我想确保烧瓶应该解析 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 验证请求正文的工作示例，将验证的数据转换回 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)

希望这有助于让您到达目的地。