ms.shahzeidi/sakhteman
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(project):change mongo

Browse Source
This commit is contained in:
shahzeydyfah
2025-01-10 22:02:00 +03:30
parent a83d979a54
commit 0630f04690
5 changed files with 292 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/utils/te/engines/Mssql.py
+112 -11
View File
@@ -1,8 +1,18 @@
from sqlalchemy import or_, and_, desc, asc
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 ..utils.Models import FilterItem, FilterScope
# from .Mongo import zero_time_data
from ..utils.Models import FilterItem, FilterScope, OutPutType
class MsSqlEngine(AbstractEngineClass):
@@ -16,7 +26,8 @@ class MsSqlEngine(AbstractEngineClass):
async def exe(self):
_f = self.set_filters()
if not self.custom_entities:
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)
@@ -29,9 +40,11 @@ class MsSqlEngine(AbstractEngineClass):
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(), tmp
return self.rows, tmp.count()
def query_generator(self):
if not self.input.has_filter():
@@ -42,8 +55,9 @@ class MsSqlEngine(AbstractEngineClass):
def set_sort(self):
if self.input.o.field:
self.sort = asc(getattr(self.model, self.input.o.field)) if self.input.o.direction == 'asc' else desc(
getattr(self.model, 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
@@ -65,9 +79,13 @@ class MsSqlEngine(AbstractEngineClass):
if has_join:
pass
else:
sm = self.model.c if isinstance(self.model, TableValuedAlias) else self.model
items.append(
getattr(self.model, col).ilike("%{search}%".format(search=self.input.get_text_search())))
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)
@@ -88,7 +106,7 @@ class MsSqlEngine(AbstractEngineClass):
"n_eq": "__ne__",
"gt": "__gt__",
"gte": "__ge__",
"lte": "__lte__",
"lte": "__le__",
"lt": "__lt__",
"like": "ilike",
"n_like": "notlike",
@@ -115,10 +133,22 @@ class MsSqlEngine(AbstractEngineClass):
def filter_item_generator(self, item: FilterItem):
has_join = len(item.col.split('.')) > 1
if has_join:
pass
# pass
return self.join_filter(item)
else:
return getattr(getattr(self.model, item.col), self.get_operator(item.op))(item.val)
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:
@@ -144,3 +174,74 @@ class MsSqlEngine(AbstractEngineClass):
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