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