table_check.py
2.2 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
#author: 4N
#createtime: 2021/1/27
#email: nheweijun@sina.com
from ..models import Table,DES,Database
from app.util.component.ApiTemplate import ApiTemplate
from app.util.component.PGUtil import PGUtil
from osgeo.ogr import Layer
class Api(ApiTemplate):
api_name = "检查表是否存在"
def process(self):
res = {}
pg_ds = None
try:
table_names = self.para.get("table_names")
database_guid = self.para.get("database_guid")
res["data"] = []
database = Database.query.filter_by(guid=database_guid).one_or_none()
if not database:
raise Exception("数据源不存在！")
pg_ds = PGUtil.open_pg_data_source(0, DES.decode(database.sqlalchemy_uri))
if not pg_ds:
raise Exception("数据源连接失败！")
for table_name in table_names.split(","):
layer:Layer = pg_ds.GetLayerByName(table_name)
table = Table.query.filter_by(database_guid=database_guid,name=table_name).one_or_none()
if table and layer:
exist = True
table_guid = table.guid
else:
exist = False
table_guid = None
res["data"].append({"table_name": table_name, "exist": exist,"table_guid":table_guid})
res["result"] = True
except Exception as e:
raise e
finally:
if pg_ds:
try:
pg_ds.Destroy()
except:
pass
return res
api_doc={
"tags":["管理接口"],
"parameters":[
{"name": "table_names",
"in": "formData",
"type": "string",
"description": "表table_names,多个用,相隔","required":"true"},
{"name": "database_guid",
"in": "formData",
"type": "string",
"description": "database guid", "required": "true"},
],
"responses":{
200:{
"schema":{
"properties":{
}
}
}
}
}