Я пришел из Java и сейчас пробую свои силы в Python.
Я пытаюсь создать веб-приложение RESTful для операций CRUD.
Я уже пытался просмотреть различные ссылки, которые обеспечивают пошаговый подход к достижению этого с помощью 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)
Когда я пытаюсь вызвать url curl -X GET "http://127.0.0.1:5000/edslnk/rest/eds-workflows/" -H" принять: application / json "я получаю код ошибки сервера 500
Моя общая задача на данном этапе - заполнить мою модель workFlow некоторыми жестко закодированными данными, а затем вернуть список этих моделей в виде данных json обратно вызывающей стороне.
Как только я смогу достичь этого, я попытаюсь использовать реальные данные из базы данных, повторяя набор результатов и заполняя строки набора результатов в модель.
Заранее спасибо за любую помощь!
********** ОБНОВЛЕНИЕ - 1 ***********
Мне удалось добиться некоторого прорыва на основе приведенного ниже кода. Я получаю ожидаемый вывод с кодом ниже.
Однако теперь я заканчиваю тем, что создаю два определения моего класса Model: одно через api.model, а другое через создание обычного class
Объект модели workFlow, созданный через api.Model, используется только для операции маршаллинга (в декораторе @ns.marshal_list_with), в то время как обычные class созданные объекты модели workFlow1 и workFlow2 используются для заполнения фактических данных и добавления их к workFlow_list
Разве невозможно получить разные экземпляры моего объекта модели с помощью api.model и заполнить их моими данными (вроде как с помощью «сеттеров», как в java)?
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)