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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
| import datetime
import flask_sqlalchemy
from easydubbo import db
ignore_filed_list = ['__class__', '__delattr__', '__dict__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__le__', '__lt__', '__mapper__', '__module__', '__ne__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', '__table__', '__tablename__', '__weakref__', '_decl_class_registry', '_sa_class_manager', '_sa_instance_state', 'check_password', 'generate_auth_token', 'get_id', 'hash_password', 'is_active', 'is_anonymous', 'is_authenticated', 'metadata', 'password', 'query', 'query_class', 'verify_auth_token']
class ModelOperate:
def __init__(self, nedd_paginate=True):
self._query_column = []
self._query_filter = []
self._query_order = []
self._query_set = ''
self.nedd_paginate = nedd_paginate
self._return_column = []
def excute_query(self):
"""
结果集，此处组装完成后还只是sql
"""
self._query_set = db.session.query(*self._query_column).filter(*self._query_filter).order_by(*self._query_order)
def set_query_column(self, *args):
"""
设置查询条件，若无，则返回所有列
"""
if not isinstance(args[0], flask_sqlalchemy.model.DefaultMeta):
self._return_column = args
self._query_column = args
def set_query_filter(self, *args):
"""
设置过滤条件，可不设置
"""
self._query_filter = args[0]
def set_query_order(self, *args):
"""
设置排序字段，可不设置
-xxx降序
+xxx升序
"""
self._query_order = args[0]
@staticmethod
def is_basequery(obj):
"""
判断是否是flask_sqlalchemy.BaseQuery类型
"""
if isinstance(obj, flask_sqlalchemy.BaseQuery):
return True
@property
def total(self):
"""
使用query set的原生方式快速获取结果集的总数
"""
if self.is_basequery(self._query_set):
return self._query_set.count()
raise Exception('Error')
def get_result(self, page=1, page_size=10):
"""
快速进行分页
"""
self.excute_query()
if self.is_basequery(self._query_set):
if self.nedd_paginate:
_result = self._query_set.paginate(page=page, per_page=page_size).items
else:
_result = self._query_set.all()
try:
if self._return_column:
return self.result_convert_to_list(_result)
return self.model_query_convert_to_list(_result)
except Exception as e:
raise Exception("Error: %s" % str(e))
def result_convert_to_list(self, results):
"""
分页结果处理
"""
_resutn_column_nums = len(self._return_column)
_results = []
for _query in results:
_dict = {}
for num in range(0, _resutn_column_nums):
data = _query[num]
name = self._return_column[num].__dict__.get('key')
if isinstance(data, datetime):
_dict[name] = datetime.strftime(data, "%Y-%m-%d %H:%M:%S")
elif data is None:
_dict[name] = ''
else:
_dict[name] = data
_results.append(_dict)
return _results
@staticmethod
def model_query_convert_to_list(results):
_list = []
for i in results:
_dict = {}
for field in [x for x in dir(i) if not x in ignore_filed_list]:
data = i.__getattribute__(field)
try:
if isinstance(data, datetime):
data = datetime.strftime(data, "%Y-%m-%d %H:%M:%S")
elif data is None:
data = ''
except Exception as e:
data = str(data)
finally:
_dict[field] = data
_list.append(_dict)
return _list
|
