alicorn-dev/KintoneAppBuilder
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bugfix change_appversion Unique Error

Browse Source
This commit is contained in:
方 柏
2024-12-10 11:27:30 +09:00
parent a3df6c4b37
commit 9eec7e835d
3 changed files with 50 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/app/db/cruddb/crudbase.py
View File

@@ -100,5 +100,5 @@ class crudbase:
query = self._apply_filters(query, filters) query = self._apply_filters(query, filters)
if sort_by: if sort_by:
query = self._apply_sorting(query, sort_by, sort_order) query = self._apply_sorting(query, sort_by, sort_order)
#print(str(query)) print(str(query))
return query return query

92
backend/app/db/cruddb/dbapp.py
View File

@@ -17,8 +17,8 @@ class dbflowhistory(crudbase):
super().__init__(model=models.FlowHistory) super().__init__(model=models.FlowHistory)
def get_flows_by_appid_version(self,db: Session,domainurl:str,appid:str,version:int): def get_flows_by_appid_version(self,db: Session,domainurl:str,appid:str,version:int):
return db.execute(select(models.FlowHistory).filter(and_(models.FlowHistory.domainurl == domainurl,models.FlowHistory.appid == appid, models.FlowHistory.version == version))).scalars().all() return db.execute(super().get_by_conditions({"domainurl":domainurl,"appid":appid, "version":version})).scalars().first()
dbflowhistory = dbflowhistory() dbflowhistory = dbflowhistory()
class dbflow(crudbase): class dbflow(crudbase):
@@ -32,7 +32,7 @@ class dbflow(crudbase):
return db.execute(super().get_by_conditions({"flowid":flowid})).scalars().first() return db.execute(super().get_by_conditions({"flowid":flowid})).scalars().first()
def get_flows_by_appid(self,db: Session,domainurl:str,appid:str): def get_flows_by_appid(self,db: Session,domainurl:str,appid:str):
return db.execute(select(models.Flow).filter(and_(models.Flow.domainurl == domainurl,models.Flow.appid == appid))).scalars().all() return db.execute(select(models.Flow).filter(and_(models.Flow.domainurl == domainurl,models.Flow.appid == appid))).scalars().first()
def create_flow(self,db: Session, domainurl: str, flow: schemas.FlowIn,userid:int): def create_flow(self,db: Session, domainurl: str, flow: schemas.FlowIn,userid:int):
@@ -111,62 +111,60 @@ class dbapp(crudbase):
db.add(appversion) db.add(appversion)
db.add(db_app) db.add(db_app)
flows = dbflow.get_flows_by_appid(db,domainurl,newversion.appid) flow = dbflow.get_flows_by_appid(db,domainurl,newversion.appid)
for flow in flows: if not flow:
db_flowhistory = models.FlowHistory( return None
flowid = flow.flowid, db_flowhistory = models.FlowHistory(
appid = flow.appid, flowid = flow.flowid,
eventid = flow.eventid, appid = flow.appid,
domainurl = flow.domainurl, eventid = flow.eventid,
name = flow.name, domainurl = flow.domainurl,
content = flow.content, name = flow.name,
version = db_app.version, content = flow.content,
updateuserid = userid, version = db_app.version,
createuserid = userid updateuserid = userid,
) createuserid = userid
db.add(db_flowhistory) )
db.add(db_flowhistory)
db.commit() db.commit()
db.refresh(db_app) db.refresh(db_app)
return db_app return db_app
return None return None
def change_appversion(self,db: Session, domainurl:str,appid:str,version:int,userid:int): def change_appversion(self,db: Session, domainurl:str,appid:str,version:int,userid:int):
db_app = self.get_app(db,domainurl,appid) db_app = self.get_app(db, domainurl, appid)
db_appversion = dbappversion.get_app_by_version(db,domainurl,appid,version) if not db_app:
if db_appversion and db_app: return None
db_app.version = version db_appversion = dbappversion.get_app_by_version(db, domainurl, appid, version)
db_app.updateuserid = userid if not db_appversion:
db.add(db_app) return None
db_app.version = version
db_app.updateuserid = userid
db.add(db_app)
flows = dbflow.get_flows_by_appid(db,domainurl,appid) #select(models.Flow).filter(and_(models.Flow.domainurl == appedit.domainurl,models.App.appid == appedit.appid)) flow = dbflow.get_flows_by_appid(db, domainurl, appid)
for flow in flows: if not flow:
db.delete(flow) return None
flows = dbflowhistory.get_flows_by_appid_version(db,domainurl,appid,version) #select(models.Flow).filter(and_(models.Flow.domainurl == appedit.domainurl,models.App.appid == appedit.appid)) flowhistory = dbflowhistory.get_flows_by_appid_version(db, domainurl, appid, version)
for flow in flows: if not flowhistory:
db_flow = models.Flow( return None
flowid = flow.flowid, flow.flowid = flowhistory.flowid
appid = flow.appid, flow.eventid=flowhistory.eventid
eventid = flow.eventid, flow.name=flowhistory.name
domainurl = flow.domainurl, flow.content=flow.content
name = flow.name, flow.updateuserid=userid
content = flow.content, db.add(flow)
updateuserid = userid, db.commit()
createuserid = userid db.refresh(db_app)
) return db_app
db.add(db_flow)
db.commit()
db.refresh(db_app)
return db_app
return None
def delete_app(self,db: Session, domainurl: str,appid: str ): def delete_app(self,db: Session, domainurl: str,appid: str ):
db_app =self.get_app(db,domainurl,appid) db_app =self.get_app(db,domainurl,appid)
if db_app: if db_app:
db_flow = dbflow.get_flows_by_appid(db,domainurl,appid)
if db_flow:
db.delete(db_flow)
db.delete(db_app) db.delete(db_app)
db_flows = dbflow.get_flows_by_appid(db,domainurl,appid)
for flow in db_flows:
db.delete(flow)
db.commit() db.commit()
return db_app return db_app
return None return None

6
backend/app/tests/test_user_app.py
View File

@@ -54,8 +54,9 @@ def test_appversions_update(test_client,test_domain,test_app_id,login_user):
"appid": test_app_id "appid": test_app_id
} }
response = test_client.post("/api/apps", json=app_version,headers={"Authorization": "Bearer " + login_user}) response = test_client.post("/api/apps", json=app_version,headers={"Authorization": "Bearer " + login_user})
assert response.status_code == 200
data = response.json() data = response.json()
logging.error(data)
assert response.status_code == 200
assert "data" in data assert "data" in data
assert data["data"] is not None assert data["data"] is not None
assert data["data"]["domainurl"] == test_domain.url assert data["data"]["domainurl"] == test_domain.url
@@ -104,7 +105,8 @@ def test_appversions_change(test_client,test_domain,test_app_id,login_user):
def test_delete_app(test_client,test_app_id,login_user): def test_delete_app(test_client,test_app_id,login_user):
response = test_client.delete("/api/apps/"+ test_app_id, headers={"Authorization": "Bearer " + login_user}) response = test_client.delete("/api/apps/"+ test_app_id, headers={"Authorization": "Bearer " + login_user})
assert response.status_code == 200
data = response.json() data = response.json()
logging.error(data)
assert response.status_code == 200
assert "data" in data assert "data" in data
assert data["data"] is not None assert data["data"] is not None