248 lines
9.3 KiB
Python
248 lines
9.3 KiB
Python
import os
|
|
import string
|
|
import tempfile
|
|
import uuid
|
|
|
|
# import xlsxwriter
|
|
from celery import shared_task
|
|
from sqlalchemy import or_, and_, desc, asc, create_engine, LABEL_STYLE_TABLENAME_PLUS_COL
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.sql.elements import BooleanClauseList, BinaryExpression, UnaryExpression
|
|
from sqlalchemy.sql.selectable import TableValuedAlias
|
|
|
|
from .AbstractEngine import AbstractEngineClass
|
|
# from .Mongo import zero_time_data
|
|
from ..utils.Models import FilterItem, FilterScope, OutPutType
|
|
|
|
|
|
class MsSqlEngine(AbstractEngineClass):
|
|
|
|
def set_joins(self):
|
|
pass
|
|
|
|
filters = {}
|
|
rows = []
|
|
sort = None
|
|
|
|
async def exe(self):
|
|
_f = self.set_filters()
|
|
print(_f)
|
|
if self.custom_entities is not None:
|
|
query = self.query_class.query(self.model)
|
|
else:
|
|
query = self.query_class.query(*self.custom_entities)
|
|
if self.joins:
|
|
for j in self.joins:
|
|
if isinstance(j, list):
|
|
query = query.join(*j)
|
|
else:
|
|
query = query.join(j)
|
|
|
|
if self.options:
|
|
query = query.options(*self.options)
|
|
if self.input.e and self.input.e.ot:
|
|
self.input.e.d = self.generate_output(query, _f)
|
|
tmp = query.filter(_f).order_by(self.set_sort())
|
|
self.rows = tmp.limit(self.input.get_limit()).offset(self.input.skip()).all()
|
|
return self.rows, tmp.count()
|
|
|
|
def query_generator(self):
|
|
if not self.input.has_filter():
|
|
return None
|
|
filters = self.input.filters()
|
|
query = self.generate_scope_filter(filters, general=True)
|
|
return query
|
|
|
|
def set_sort(self):
|
|
if self.input.o.field:
|
|
sm = self.model.c if isinstance(self.model, TableValuedAlias) else self.model
|
|
self.sort = asc(getattr(sm, self.input.o.field)) if self.input.o.direction == 'asc' else desc(
|
|
getattr(sm, self.input.o.field))
|
|
return self.sort
|
|
elif isinstance(self.default_sort_by, UnaryExpression):
|
|
return self.default_sort_by
|
|
else:
|
|
return True
|
|
|
|
def set_filters(self):
|
|
filters = []
|
|
query = self.query_generator()
|
|
if isinstance(query, BooleanClauseList) or isinstance(query, BinaryExpression):
|
|
filters.append(query)
|
|
if self.input.has_text_search():
|
|
items = []
|
|
_search_cols = self.input.get_search_cols() if (
|
|
self.input.get_search_cols() and len(self.input.get_search_cols())) else self.search_cols
|
|
if _search_cols:
|
|
for col in _search_cols:
|
|
has_join = len(col.split('.')) > 1
|
|
if has_join:
|
|
pass
|
|
else:
|
|
sm = self.model.c if isinstance(self.model, TableValuedAlias) else self.model
|
|
items.append(
|
|
getattr(sm, col).ilike(
|
|
"%{search}%".format(search=self.input.get_text_search())))
|
|
|
|
filters.append(or_(*items))
|
|
|
|
if self.input.extra_filter:
|
|
for f in self.input.extra_filter:
|
|
filters.append(f)
|
|
if not filters:
|
|
return True
|
|
self.filters = and_(*filters)
|
|
return self.filters
|
|
|
|
@staticmethod
|
|
def get_expression_filter(col=None, op: str = '', val: str = '') -> dict:
|
|
_op = MsSqlEngine.get_operator(op)
|
|
return getattr(col, op)(val)
|
|
|
|
@staticmethod
|
|
def get_operator(op: str):
|
|
ops = {
|
|
"eq": "__eq__",
|
|
"n_eq": "__ne__",
|
|
"gt": "__gt__",
|
|
"gte": "__ge__",
|
|
"lte": "__le__",
|
|
"lt": "__lt__",
|
|
"like": "ilike",
|
|
"n_like": "notlike",
|
|
"in": "in_",
|
|
"n_in": "not_in",
|
|
"bet": "between",
|
|
"n_bet": "not_between",
|
|
"sw": "startswith",
|
|
"ew": "endswith",
|
|
'n_ew': 'not_endswith',
|
|
'n_sw': 'not_startswith',
|
|
'exi': 'exi',
|
|
"n_exi": "n_exi",
|
|
}
|
|
return ops[op] if op in ops else '__eq__'
|
|
|
|
@staticmethod
|
|
def get_and_or(op):
|
|
return or_ if op == 'or' else and_
|
|
|
|
def count(self):
|
|
pass
|
|
|
|
def filter_item_generator(self, item: FilterItem):
|
|
has_join = len(item.col.split('.')) > 1
|
|
if has_join:
|
|
# pass
|
|
return self.join_filter(item)
|
|
else:
|
|
sm = self.model.c if isinstance(self.model, TableValuedAlias) else self.model
|
|
# if item.type in ['DATE', 'DATE_TIME']:
|
|
# if item.op == 'eq':
|
|
# return getattr(sm, item.col).between(zero_time_data(item.val),
|
|
# zero_time_data(item.val, is_max=True))
|
|
# if item.op in ['gte', 'gt']:
|
|
# item.val = zero_time_data(item.val)
|
|
# print(item.val,'GT')
|
|
# if item.op in ['lt', 'lte']:
|
|
# item.val = zero_time_data(item.val, is_max=True)
|
|
# print(item.val,'LT')
|
|
|
|
return getattr(getattr(sm, item.col), self.get_operator(item.op))(item.val)
|
|
|
|
def generate_scope_filter(self, filters: list, general=False):
|
|
if len(filters) == 1:
|
|
_o = filters[0]
|
|
if isinstance(_o, FilterItem):
|
|
return self.filter_item_generator(filters[0])
|
|
elif isinstance(_o, FilterScope):
|
|
return self.generate_scope_filter(_o.items)
|
|
else:
|
|
return _o
|
|
else:
|
|
_a = filters.pop(0)
|
|
_a = (self.filter_item_generator(_a) if isinstance(_a,
|
|
FilterItem) else _a) if not general else (
|
|
self.generate_scope_filter(_a.items) if isinstance(_a, FilterScope) else _a)
|
|
|
|
_b = filters.pop(0)
|
|
|
|
_op = MsSqlEngine.get_and_or(_b.cr if not general else _b.op)
|
|
_b = (self.filter_item_generator(_b) if isinstance(_b,
|
|
FilterItem) else _b) if not general else (
|
|
self.generate_scope_filter(_b.items) if isinstance(_b, FilterScope) else _b)
|
|
|
|
filters = [_op(_a, _b)] + filters
|
|
return self.generate_scope_filter(filters, general)
|
|
|
|
def join_filter(self, item: FilterItem):
|
|
inner_joins = item.col.split('.')[-2:]
|
|
join = None
|
|
for self_join in self.joins:
|
|
if inner_joins[0] == self_join.__name__ or self.normalize_model_name(inner_joins[0]) == self_join.__name__:
|
|
join = self_join
|
|
return getattr(getattr(join, inner_joins[1]), self.get_operator(item.op))(item.val) if join else None
|
|
|
|
@staticmethod
|
|
def normalize_model_name(obj):
|
|
if isinstance(obj, str):
|
|
return obj.replace('_', '')
|
|
else:
|
|
result = {}
|
|
model_name = str([i for i in obj.keys()][0]).rsplit('_', 1)
|
|
model_name.pop(-1)
|
|
model_name = ''.join(model_name)
|
|
for index in obj.keys():
|
|
item = index
|
|
for i in range(5):
|
|
item = item.replace(f'_{i}_', '_')
|
|
if item[-1] == '_':
|
|
item = item[:-1]
|
|
item = item.replace(model_name + '_', '')
|
|
result[item] = obj[index]
|
|
return result
|
|
|
|
def generate_output(self, query, _f):
|
|
if self.input.e.ot == OutPutType.EXEL:
|
|
tmp = query.filter(_f).order_by(self.set_sort()).set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
|
|
query = tmp.statement.compile()
|
|
params = query.params
|
|
file_name = str(uuid.uuid4().hex) + '.xlsx'
|
|
MsSqlEngine.output_generator(str(query).replace('false', '0'), params,
|
|
str(self.query_class.bind.url), file_name=file_name, cols=self.input.e.c)
|
|
return file_name
|
|
elif self.input.e.ot == OutPutType.JSON:
|
|
pass
|
|
elif self.input.e.ot == OutPutType.PDF:
|
|
pass
|
|
else:
|
|
raise Exception("Output type not supported.")
|
|
|
|
@staticmethod
|
|
@shared_task(name='output_generator_task')
|
|
def output_generator(query, params, url, file_name, cols):
|
|
# engine = create_engine(url=url)
|
|
# session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
# db = session()
|
|
# datas = db.execute(query, params)
|
|
# file_dir = os.path.join(tempfile.gettempdir(), file_name)
|
|
# head_cols = list(str(string.ascii_uppercase))
|
|
# headers = cols.keys()
|
|
# # workbook = xlsxwriter.Workbook(file_dir)
|
|
# worksheet = workbook.add_worksheet()
|
|
# for header in headers:
|
|
# worksheet.write(f'{head_cols.pop(0)}1', cols[header])
|
|
# row = 1
|
|
# for item in datas:
|
|
# row += 1
|
|
# head_cols = list(str(string.ascii_uppercase))
|
|
# data = MsSqlEngine.normalize_model_name(item)
|
|
# for col in headers:
|
|
# if col[-1] == '_':
|
|
# col = col[:-1]
|
|
# col = col.replace('.', '_')
|
|
# worksheet.write(f'{head_cols.pop(0)}{row}', data[col] if col in data else '-')
|
|
# workbook.close()
|
|
...
|
|
return True
|