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

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)的函数。

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

事实上，我想我可以很容易地在route函数中调用这个方法，但它看起来很奇怪，而且可能还有其他我找不到的方法，也许是decorator？

Answer 1

重构的一种可能性是使用webargs、Flask Marshmallow和棉花糖sqlalchemy。

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

以下示例基于您的信息，并简要概述了使用情况。通过定义自己的错误处理程序，错误消息也可以作为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()