ModelVisitor.py
2.8 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
79
80
81
82
83
84
85
86
87
88
89
# coding=utf-8
#author: 4N
#createtime: 2021/5/17
#email: nheweijun@sina.com
from .PGUtil import PGUtil
from app.models import DES
import datetime
from flask_sqlalchemy import SQLAlchemy,Model
class ModelVisitor:
@classmethod
def to_json(cls,model):
pass
@classmethod
def object_to_json(cls,obj):
info={}
if obj:
info = cls.formatter(obj.__dict__)
return info
@classmethod
def formatter(cls,kvdict:dict):
def generator():
for k,value in kvdict.items():
if isinstance(value, Model) or k.__eq__('_sa_instance_state'):
continue
if isinstance(value, datetime.datetime):
value = value.strftime('%Y-%m-%d %H:%M:%S')
elif isinstance(value, datetime.time):
value = value.strftime('%H:%M:%S')
yield k,value
return dict(generator())
@classmethod
def objects_to_jsonarray(cls,objects):
result = []
for o in objects:
result.append(cls.object_to_json(o))
return result
@classmethod
def table_to_json(cls,table):
info = {}
info["catalog_name"] = table.relate_catalog.name if table.relate_catalog else None
info["database_alias"] = table.relate_database.alias if table.relate_database else None
if table:
info.update(cls.formatter(table.__dict__))
return info
@classmethod
def task_to_json(cls,task):
info = {}
info["catalog_name"] = task.relate_catalog.name if task.relate_catalog else None
info["database_alias"] = task.relate_database.alias if task.relate_database else None
info["consume_time"]=None
if task.update_time:
consume_time = int(task.update_time.timestamp()-task.create_time.timestamp())
minute = int(consume_time/60)
sec = 1 if (int(consume_time%60)==0 and minute==0) else int(consume_time%60)
if minute>0:
info["consume_time"] = "{}分{}秒".format(minute,sec)
else:
info["consume_time"] = "{}秒".format(sec)
if object:
info.update(cls.formatter(task.__dict__))
del info["parameter"]
return info
@classmethod
def database_to_json(cls,database):
info = {}
if object:
info.update(cls.formatter(database.__dict__))
user, password, host, port, db= PGUtil.get_info_from_sqlachemy_uri(DES.decode(info["sqlalchemy_uri"]))
info["user"] = user
info["host"] = host
info["port"] = port
info["database"] = db
del info["sqlalchemy_uri"]
return info