时间戳

首页 > 解决方案 > flask：如何对请求 JSON 和 JSON 模式进行验证？

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

问题描述

在 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？

标签： pythonjsonflask

解决方案

正如我试图在我们的谈话中传达的那样，您似乎正在使用序列化和反序列化工具。我发现 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)

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

推荐阅读