Initial commit

core/utils/te/utils/Encoder.py

@@ -0,0 +1,19 @@
+import json
+
+from sqlalchemy.orm import DeclarativeMeta
+
+
+class AlchemyEncoder(json.JSONEncoder):
+    def default(self, obj):
+        if isinstance(obj.__class__, DeclarativeMeta):
+            fields = {}
+            for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:
+                data = obj.__getattribute__(field)
+                try:
+                    json.dumps(data)
+                    fields[field] = data
+                except TypeError:
+                    fields[field] = None
+            return fields
+
+        return json.JSONEncoder.default(self, obj)

core/utils/te/utils/Models.py

@@ -0,0 +1,160 @@
+from enum import Enum
+from typing import Optional, Any, List
+
+import pydantic
+from pydantic import BaseModel
+
+
+class FilterItem(BaseModel):
+    col: Optional[str]
+    cr: Optional[str]
+    id: Optional[str]
+    op: Optional[str]
+    type: Optional[str]
+    val: Optional[Any]
+
+    @staticmethod
+    def ValidItem(item: dict):
+        return item.get('col') and \
+            item.get('id') and \
+            item.get('op') and \
+            item.get('type')
+
+    @staticmethod
+    def ValidValue(item: dict):
+        return len(str(item.get('val', '')).strip()) > 0
+
+
+class Cr(str, Enum):
+    OR = 'or'
+    AND = 'and'
+
+
+class FilterScope(BaseModel):
+    id: str
+    items: Optional[List[FilterItem]]
+    op: Optional[Cr]
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        if kwargs.get('items'):
+            self.items = [
+                FilterItem(**item) for item in filter(lambda i: FilterItem.ValidItem(i), kwargs.get('items', []))]
+        self.op = Cr(kwargs.get('op'))
+        self.id = str(kwargs.get('id'))
+
+    @staticmethod
+    def ValidScope(item: dict):
+        return item.get('id') and item.get('items') and len(
+            item.get('items')) > 0 and item.get('op')
+
+    @staticmethod
+    def CleanItems(item):
+        clone = item.dict()
+        clone['items'] = list(filter(
+            lambda i: FilterItem.ValidValue(i), clone.get(
+                'items', [])))
+        return FilterScope(**clone)
+
+
+class Page(BaseModel):
+    c: int = 1
+    s: int = 5
+
+
+class SearchModel(BaseModel):
+    text: Optional[str]
+    cols: Optional[List[Any]]
+
+
+class SortModel(BaseModel):
+    field: Optional[str]
+    direction: Optional[str]
+
+
+class JoinModel(BaseModel):
+    model: BaseModel
+    show: List[str]
+    search: Optional[List[str]]
+
+
+class Config:
+    extra = pydantic.Extra.ignore
+    any_str_strip_whitespace = True  # strip whitespaces from strings
+    underscore_attrs_are_private = False
+    arbitrary_types_allowed = True
+
+
+class TableEngineInput(BaseModel):
+    f: Optional[List[FilterScope]]
+    extra_filter: list = []
+    o: Optional[SortModel]
+    p: Optional[Page] = Page()
+    s: Optional[SearchModel]
+    is_deleted: Optional[bool]
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        _input = {**{'p': {'c': 1, 's': 5},
+                     'f': [], 's': {}, 'o': {}}, **(kwargs or {})}
+        self.p = Page(**_input.get('p'))
+        self.o = SortModel(**_input.get('o'))
+        if _input.get('f'):
+            self.f = [FilterScope(**f) for f in
+                      filter(lambda fs: FilterScope.ValidScope(fs), _input.get('f'))]
+            self.f = list(map(lambda fs: FilterScope.CleanItems(fs), self.f))
+        if _input.get('s'):
+            self.s = SearchModel(**_input.get('s'))
+
+    def limit(self):
+        return self.p.s
+
+    def order(self):
+        return self.o
+
+    def filters(self) -> List[FilterScope]:
+        return self.f
+
+    def has_filter(self) -> bool:
+        return self.f is not None and \
+            len(self.f) > 0 and \
+            self.f[0].items is not None and \
+            len(self.f[0].items) > 0
+
+    def has_extra_filter(self) -> bool:
+        return self.extra_filter is not None and \
+            len(self.extra_filter) > 0
+
+    def skip(self):
+        return self.page() * self.p.s
+
+    def page(self):
+        return self.p.c - 1
+
+    def virtual_page(self):
+        return self.page() + 1
+
+    def get_limit(self):
+        limit = self.p.s if self.p else 20
+        return limit if limit < 100 else 100
+
+    def text_search_cols(self, cols: List[str]):
+        if self.s:
+            self.s.cols = cols
+
+    def has_text_search(self) -> bool:
+        return self.s is not None and self.s.text is not None and len(self.s.text.strip()) > 0
+
+    def add_filter(self, _filter):
+        self.extra_filter.append(_filter)
+
+    def get_search_cols(self) -> List[str]:
+        return self.s.cols if self.s else []
+
+    def get_text_search(self) -> str:
+        return self.s.text.strip() if self.s else None
+
+
+class WorkingMode(Enum):
+    Development = "development"
+    Production = "production"