如何使用python Flask RESTplus填充模型

Question

我来自Java背景，现在正在尝试使用Python。 我正在尝试为 CRUD 操作创建一个 RESTful Web 应用程序。 我已经尝试通过各种链接，这些链接提供了使用Flask RESTplus + SQLAlchemy逐步实现它的方法。

但是，我更热衷于以通常的"java方式"获得它。我还想避免使用 ORM，而是使用 SQLAlchemy 执行原始 SQL，然后迭代结果集以填充我的模型，然后返回模型。

我正在采取婴儿步骤，最初我想看看如何使用虚拟数据填充我的模型并将其返回给调用方。以下是我正在努力实现但无法进一步实现的目标：

from flask import request
from flask import Flask
from flask import Blueprint
from flask_restplus import Resource, Api, apidoc
from flask_restplus import fields, marshal
from sqlalchemy import create_engine
from sqlalchemy.sql import text
import json
app = Flask(__name__)
blueprint = Blueprint('api', __name__, url_prefix='/edslnk/rest')
api = Api(blueprint, version='1.0', title='EDSLink ResourceSvc API',description='Set of APIs to interactively communicate with Database and HDFS')
app.register_blueprint(blueprint)
ns = api.namespace('eds-workflows', description='Operations related to EDS Workflows')
#engine = create_engine('oracle+cx_oracle://scott:tiger@tnsname')
workFlow = api.model('WorkFlow', {
'workflowName': fields.String,
'workflowScriptLoc': fields.String,
'createUser': fields.String,
'createDate': fields.String,
'lstUpdateUser': fields.String,
'lstUpdateDate': fields.String,
})
@ns.route('/')
class WorkflowsCollection(Resource):
@ns.marshal_list_with(workFlow)
def get(self):
try:
#connection = engine.connect()
#result = connection.execute("select * from EDSOPERATION.EDSLNK_WORKFLOWS")
workFlow_list = []
for x in range(6):
workFlow.workflowName = 'Workflow_Name'+x
workFlow.workflowScriptLoc = 'Workflow_Script_Loc'+x
workFlow.createUser = 'Create_User'+x
workFlow.createDate = 'Create_Date'+x
workFlow.lstUpdateUser = 'Lst_Modified_User'+x
workFlow.lstUpdateDate = 'Lst_Modified_Date'+x
workFlow_list.append(workFlow)
finally:
#connection.close()
print('finally')
#what should I return ? 'workFlow' or 'workFlow_list' ?

@ns.response(201, 'Category successfully created.')
def post(self):
#create_workflow(request.json)
pass
if __name__ == '__main__':
app.run(debug=True)

当我尝试调用网址curl -X GET "http://127.0.0.1:5000/edslnk/rest/eds-workflows/" -H "accept： application/json" 时，我得到服务器错误代码 500

我在此阶段的总体目标是使用一些硬编码数据填充我的模型workFlow，然后将这些模型的列表作为 json 数据返回给调用方。

一旦我能够实现这一点，我将尝试通过迭代结果集并将结果集行填充到模型中来使用数据库中的实际数据。

提前感谢任何帮助！

更新 - 1***********

我能够根据下面的代码实现一些突破。我通过以下代码获得预期的输出。

但是，我现在最终创建了模型类的两个定义，即一个通过api.model，另一个通过创建普通class模型对象workFlow通过 API 创建的。模型仅用于编组操作(在@ns.marshal_list_with装饰器中(，而普通class创建的模型对象workFlow1和workFlow2用于填充实际数据并将它们添加到workFlow_list

是否无法使用api.model获取我的模型对象的不同实例并用我的数据填充它们(有点像在 java 中使用"setters"(？

from flask import request
from flask import Flask,jsonify
from flask import Blueprint
from flask_restplus import Resource, Api, apidoc
from flask_restplus import fields, marshal
from sqlalchemy import create_engine
from sqlalchemy.sql import text
import json
app = Flask(__name__)
blueprint = Blueprint('api', __name__, url_prefix='/edslnk/rest')
api = Api(blueprint, version='1.0', title='EDSLink ResourceSvc API',description='Set of APIs to interactively communicate with Database and HDFS')
app.register_blueprint(blueprint)
ns = api.namespace('eds-workflows', description='Operations related to EDS Workflows')
#engine = create_engine('oracle+cx_oracle://scott:tiger@tnsname')
workFlow = api.model('WorkFlow', {
'workflowName': fields.String,
'workflowScriptLoc': fields.String,
'createUser': fields.String,
'createDate': fields.String,
'lstUpdateUser': fields.String,
'lstUpdateDate': fields.String,
})
#workFlow_list = []
class WorkFlow:
def __init__(self, workflowName, workflowScriptLoc, createUser, createDate, lstUpdateUser, lstUpdateDate):
self.workflowName = workflowName
self.workflowScriptLoc = workflowScriptLoc
self.createUser = createUser
self.createDate = createDate
self.lstUpdateUser = lstUpdateUser
self.lstUpdateDate = lstUpdateDate
class MyEncoder(json.JSONEncoder):
def default(self, o):
return o.__dict__

@ns.route('/')
class WorkflowsCollection(Resource):
@ns.marshal_list_with(workFlow)
def get(self):
try:
#connection = engine.connect()
#result = connection.execute("select * from EDSOPERATION.EDSLNK_WORKFLOWS")
workFlow_list = []
#workFlow1 = WorkFlow()
#workFlow1.workflowName='abc'
workFlow1 = WorkFlow('Workflow_Name', 'Workflow_Script_Loc', 'Create_User', 'Create_Date', 'Lst_Modified_User', 'Lst_Modified_Date')
workFlow2 = WorkFlow('Workflow_Name', 'Workflow_Script_Loc', 'Create_User', 'Create_Date', 'Lst_Modified_User', 'Lst_Modified_Date')
workFlow_list.append(workFlow1)
workFlow_list.append(workFlow2)
#responseString = json.dumps(workFlow_list, cls=MyEncoder)
#print('responseString --> ' + responseString)
#response = app.response_class(response=workFlow_list,status=200,mimetype='application/json')
#           for x in range(6):
#                workFlow.workflowName = 'Workflow_Name'+x
#                workFlow.workflowScriptLoc = 'Workflow_Script_Loc'+x
#                workFlow.createUser = 'Create_User'+x
#                workFlow.createDate = 'Create_Date'+x
#                workFlow.lstUpdateUser = 'Lst_Modified_User'+x
#                workFlow.lstUpdateDate = 'Lst_Modified_Date'+x
#                workFlow_list.append(workFlow)
finally:
#connection.close()
print('finally')
return workFlow_list
@ns.response(201, 'Category successfully created.')
def post(self):
#create_workflow(request.json)
pass
if __name__ == '__main__':
#x = WorkflowsCollection()
#r = x.get()
#print(r)
app.run(debug=True)

Answer 1

Flask restful 模型不是适合您尝试执行的操作的工具。它们旨在作为输入验证模型和输出序列化程序。为了验证输入，您可以使用装饰器@ns.expect(your_model)它将检查您的有效负载。

而是创建一个简单的类，就像您在更新中为尝试建模的对象所做的类一样，并创建该对象的实例数组。如果从资源获取方法返回该列表并使用@marshal_list_with(your_model)装饰器，则将获得所需的输出。

至于在两个地方定义的模型，api.model模型充当内部模型与API调用的每个字段或其结构之间的接口。但是，如果使用棉花糖，则可以从sqlalchemy模型创建API模型，marshmallow是一个更通用的验证/序列化库，这是restplus打算在未来使用的。

此外，我建议您远离处理连接，特别是不要创建SQL语句来操作对象。Sqlalchemy是一个非常富有表现力的orm，如果不使用它，您将错过很多。