flask: как сделать проверку по схеме Request JSON и JSON?

Question

В flask -restplus API мне нужно выполнить проверку данных запроса JSON, где я уже определил схему тела запроса с помощью api.model. В основном я хочу передать входные данные JSON в функцию API, где мне нужно проверить входные данные JSON перед использованием функции API. Для этого я использовал 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"
    }
  }
}

моя текущая попытка в flask

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 тела, после проверки я хочу построить JSON тело, которое будет использоваться в качестве параметра для функции API. Как я могу это сделать? какой обходной путь для этого?

проанализировано JSON должно пройти к my_funcs

в моем сообщении, данные запроса JSON должны быть проанализированы, например age, gender должны быть проверены как ожидаемые аргументы в запросе JSON, затем jsonify добавленные аргументы как JSON и передают my_funcs. как сделать это легко в fl

Я хочу убедиться, что flask должен анализировать тело JSON и добавлять аргументы, как ожидалось, иначе возникнет ошибка. например:

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

, если я предоставлю JSON данные, как указано выше, для выполнения запроса POST к конечной точке сервера, он должен выдать ошибку. Как этого добиться? как проверить запрос POST JSON для flask вызова API?

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)

Надеюсь, это поможет вам добраться туда, куда вы собираетесь.