python - 如何构建 JSON 架构以验证 DynamoDB 和 REST API？

Question

我正在编写一个 REST API，它将多个复杂对象存储到 AWS DynamoDB，然后根据请求检索它们、对它们执行计算并返回结果。这是大量提取、简化、重命名的伪代码。

class Widget:
    def __init__(self, height, weight):
        self.height = height
        self.weight = weight

class Machine:
    def __init__ (self, widgets):
        self.widgets = widgets
    def useful_method ():
        return "something great"

class WidgetSchema (Schema):
    height = fields.Decimal()
    weight = fields.Decimal()
    @post_load
    def make_widget (self, data):
        return Widget(*data)

class MachineSchema (Schema):
    widgets = fields.List(fields.Nested(WidgetSchema))
    def make_machine (self, data):
        return Machine(*data)

app = Flask(__name__)
dynamodb = boto3.resource("dynamodb", ...) 

@app.route("/machine/<uuid:machine_id>", methods=['POST'])
def create_machine(machine_id):
    input_json = request.get_json()
    validated_input = MachineSchema().load(input_json)
    # NOTE: validated_input should be a Python dict which
    # contains Decimals instead of floats, for storage in DynamoDB.
    validate_input['id'] = machine_id
    dynamodb.Table('machine').put_item(Item=validate_input)
    return jsonify({"status", "success", error_message = ""})

@app.route("/machine/<uuid:machine_id>/compute", methods=['GET'])
def get_machine(machine_id):
    result = dynamodb.Table('machine').get_item(Key=machine_id)
    return jsonify(result['Item'])

@app.route("/machine/<uuid:machine_id>/compute", methods=['GET'])
def compute_machine(machine_id):
    result = dynamodb.Table('machine').get_item(Key=machine_id)
    validated_input = MachineSchema().load(result['Item'])
    # NOTE: validated_input should be a Machine object
    # which has made use of the post_load
    return jsonify(validated_input.useful_method())

问题是我需要让我的棉花糖模式承担双重职责。对于初学者来说，在 create_machine 函数中，我需要架构来确保调用我的 REST API 的用户向我传递了一个正确形成的对象，没有额外的字段并且满足所有必需的字段等。我需要确保我没有存储毕竟数据库中存在无效垃圾。它还需要递归地抓取输入 JSON 并将所有 JSON 值转换为正确的类型。例如，Dynamo 不支持 float ，因此它们需要是小数，如下所示。这是棉花糖做的很容易的事情。如果没有 post_load，这正是将生成的 valid_input。

该架构的第二个任务是它需要获取从 DynamoDB 检索的 Python 对象，该对象看起来几乎与用户输入 JSON 完全一样，除了浮点是小数之外，并且转换将其放入我的 Python 对象、Machine 和 Widget 中。这是我需要再次读取对象的地方，但这次使用后加载来创建对象。然而，在这种情况下，我不希望我的数字是小数。我希望它们成为标准的 Python float 。

我可以为此编写两个完全不同的棉花糖模式并清楚地完成它。一种是用小数表示高度和体重，一种是用 float 。一个可以对每个对象进行后期加载，有些则没有。但编写两个相同的模式是一个巨大的痛苦。我的模式定义有几百行长。通过后加载继承数据库版本似乎不是正确的方向，因为我需要更改任何字段。嵌套以指向正确的类。例如，即使我从 MachineSchema 继承了 MachineSchemaDBVersion，并添加了 post_load，MachineScehemaDBVersion 仍然会引用 WidgetScehema，而不是 WidgetSchema 的某些数据库版本，除非我也覆盖了 widgets 字段。

我可能会派生自己的 Schema 对象，并传递一个标志来判断我们是否处于数据库模式。

人们通常如何处理希望将 REST API 输入或多或少直接存储到 DynamoDB 并进行一些验证，然后使用该数据构建 Python 对象进行计算的问题？

我尝试过的方法是让我的架构始终实例化我的 Python 对象，然后使用完全构造的对象中的转储将它们哑巴到数据库中。问题在于计算库的对象(在我的示例 Machine 或 Widget 中)不具有我需要存储在数据库中的所有必需字段，例如 ID、名称或描述。这些对象是专门为进行计算而创建的。

Answer 1

我最终找到了解决这个问题的方法。实际上，我所做的就是生成专门用于从 DynamoDB 转换为 Python 对象的 Marshmallow 架构。所有 Schema 类都具有转换为 Python 对象的 @post_load 方法，并且所有字段都标有它们在 Python 世界(而不是数据库世界)中所需的类型。

当验证来自 REST API 的输入并确保不允许错误数据进入数据库时​​，我调用 MySchema().validate(input_json)，检查是否存在错误，如果没有，则将 input_json 转储到数据库中。

这只剩下一个额外的问题，即需要清理 input_json 才能进入数据库，这是我之前使用 Marshmallow 所做的。不过，这也可以通过调整我的 JSON 解码器以从 float 读取小数来轻松完成。

总而言之，我的 JSON 解码器正在递归地遍历数据结构并将 Float 转换为 Decimal，与 Marshmallow 分开。 Marshmallow 对每个对象的字段运行验证，但仅检查结果是否有错误。然后原始输入被转储到数据库中。

我需要添加这一行来转换为十进制。

app.json_decoder = partial(flask.json.JSONDecoder, parse_float=decimal.Decimal)

我的创建函数现在看起来像这样。请注意，由我更新的 JSON 解码器解析的原始 input_json 是如何直接插入到数据库中的，而不是从 Marshmallow 输出的任何数据。

@app.route("/machine/<uuid:machine_id>", methods=['POST'])
def create_machine(machine_id):
    input_json = request.get_json() # Already ready to be DB input as is.
    errors = MachineSchema().validate(input_json)
    if errors:
      return jsonify({"status": "failure",message = dumps(errors)})
    else:
      input_json['id'] = machine_id
      dynamodb.Table('machine').put_item(Item=input_json)
      return jsonify({"status", "success", error_message = ""})