在路由方法中调用函数的最佳实践

Question

我是 Flask 新手，为了重构 Flask API 上的现有路由方法，我正在寻找减少它并在路由方法中调用方法的最佳实践。

实际上路线是这样设计的：

@qman.route('/add_report/', methods=['POST'])
def create_report():
   """
   Check if data send throught http POST request, is correct based on the report
   schema and not already recorded in the table report of the DB.
   :param: data from POST request
   :return: Ok, valide and imported -> 201, Correct but AlreadyKnown -> 208, 
   InvalideScheme -> 422
   """
   jsonData = request.get_json()
   reportSchema = ReportSchema()  
   try:
      data = reportSchema.load(jsonData)
   except ValidationError as validation_err:
      return(validation_err.messages), 422
    
   nameReportCheck = data["report_name"]
   report = Report.query.filter_by(report_name=nameReportCheck).first()
   
   if report is None:
      # Create new report
      report = Report(
      report_name=nameReportCheck,
      hostname=data["hostname"],
      status=data["status"],
      date=data["date"],
      nb_analysis_erreur=data["nb_analysis_erreur"]
      )

      db.session.add(report)
      db.session.commit()

      NewResult = reportSchema.dump(Report.query.get(report.reportID))
      return{"message" : "Created new report" , "report" : NewResult}, 201

   else :
      reportAlreadyKnown = reportSchema.dump(Report.query.get(report.reportID))
      return{"message" : "This report is already in the DB", "report" : reportAlreadyKnown}, 208

事实上，我想调用一个名为valid_schema(_schema, _jsondata) 的函数来检查通过POST 请求发送的数据是否与我的模型Report() 的架构匹配。

这个函数返回一个带有序列化数据的 Response() 对象和一个 200 代码，如果它可以序列化或者我在 try/except 中包含 400 错误代码的错误。

def valid_schema(_schema, _jsondata):
    schema = _schema()
    try:
      data = schema.load(_jsondata)
    except ValidationError as validation_err:
        response = Response(validation_err.messages, 422)
        return response
    response = Response(data, 200, mimetype="application/json")
    return response

如果valid_schema(_schema, _jsondata)在响应对象中返回report_data和200代码，则路由方法调用另一个名为create_report(report_data)的函数。

使用他的参数，这个方法检查记录是否已经在数据库中，如果不是，他从report_data arg 创建一个Report() 对象并将这个作为新记录插入到数据库中。

事实上我想我可以很容易地在路由函数中调用这个方法，但它看起来很奇怪，可能还有另一种我找不到的方法，也许是装饰器？

Answer 1

重构的一种可能性是使用webargs、Flask-Marshmallow 和marshmallow-sqlalchemy。

使用 Flask-Marshmallow，您可以通过指定字段和验证器来检查输入。 Webargs 为您提供了在装饰器中验证定义的方案并将其作为参数传递给路由的选项。结合使用marshmallow-sqlalchemy，立即转换为数据库模型。

以下示例基于您的信息，并为您简要介绍了用法。通过定义您自己的error handler，错误消息也可以作为 JSON 发送。可以在蓝图、视图等中使用。

from flask import Flask
from flask import jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from marshmallow.validate import Length, OneOf
from webargs.flaskparser import use_args

app = Flask(__name__)
db = SQLAlchemy(app)
ma = Marshmallow(app)

class Report(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String, unique=True)
    hostname = db.Column(db.String)
    status = db.Column(db.String)
    date = db.Column(db.DateTime)
    nb_analysis_error = db.Column(db.String)

class ReportSchema(ma.SQLAlchemyAutoSchema):
    class Meta:
        model = Report
        load_instance = True
        sqla_session = db.session

    name = ma.Str(required=True, validate=Length(min=3))
    hostname = ma.Str(required=True)
    date = ma.DateTime(required=True)
    status = ma.Str(required=True, validate=OneOf(['online', 'offline']))
    nb_analysis_error = ma.Str(missing='Unknown Error')

@app.route('/add_report', methods=['POST'])
@use_args(ReportSchema(), location='json')
def add_report(report):
    report_schema = ReportSchema()
    _report = Report.query.filter_by(name=report.name).first()
    if _report:
        report_data = report_schema.dump(_report)
        return jsonify(message='Already Reported', report=report_data), 208
    else:
        db.session.add(report)
        db.session.commit()
        report_data = report_schema.dump(report)
        return jsonify(message='Created', report=report_data), 201

with app.app_context():
    db.drop_all()
    db.create_all()