database_delete.py
2.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
# coding=utf-8
#author: 4N
#createtime: 2021/3/9
#email: nheweijun@sina.com
from ..models import Database,db,Table,TableVacuate,DES
from app.util.component.ApiTemplate import ApiTemplate
from app.util.component.PGUtil import PGUtil
import configure
from osgeo.ogr import DataSource
from flask import current_app
from app.util.component.UserCheck import UserCheck
class Api(ApiTemplate):
api_name = "删除数据库"
def process(self):
re ={}
va_ds = None
try:
database = db.session.query(Database).filter_by(guid=self.para.get("guid")).one_or_none()
#验证权限
UserCheck.verify(database.creator)
if configure.VACUATE_DB_URI:
va_ds: DataSource = PGUtil.open_pg_data_source(1, configure.VACUATE_DB_URI)
else:
va_ds: DataSource = PGUtil.open_pg_data_source(1, DES.decode(database.sqlalchemy_uri))
if database:
# 删除table
re_tables = Table.query.filter_by(database_guid=database.guid).all()
for table in re_tables:
re_tvs = TableVacuate.query.filter_by(table_guid=table.guid).all()
for tv in re_tvs:
db.session.delete(tv)
try:
va_ds.DeleteLayer(tv.name)
except:
current_app.logger.warning("{}不存在".format(tv.name))
db.session.delete(table)
db.session.delete(database)
db.session.commit()
re["result"] = True
re["msg"] = "数据库删除成功！"
else:
re["result"] = False
re["msg"] = "数据库不存在！"
except Exception as e:
db.session.rollback()
va_ds.RollbackTransaction()
raise e
finally:
if va_ds:
va_ds.Destroy()
return re
api_doc={
"tags":["数据库接口"],
"parameters":[
{"name": "guid",
"in": "formData",
"type": "string","description":"数据库guid","required": "true"}
],
"responses":{
200:{
"schema":{
"properties":{
}
}
}
}
}