160 lines
4.3 KiB
Python
160 lines
4.3 KiB
Python
from enum import Enum
|
|
from typing import List, Optional, Any
|
|
|
|
import pydantic
|
|
from pydantic import BaseModel
|
|
|
|
|
|
class Cr(str, Enum):
|
|
OR = 'or'
|
|
AND = 'and'
|
|
|
|
|
|
class FilterItem(BaseModel):
|
|
col: Optional[str]
|
|
cr: Optional[Cr]
|
|
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 TableEngineInput(BaseModel):
|
|
|
|
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 TableEngineInput.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 Config:
|
|
extra = pydantic.Extra.ignore
|
|
anystr_strip_whitespace = True # strip whitespaces from strings
|
|
underscore_attrs_are_private = False
|
|
arbitrary_types_allowed = True
|
|
|
|
f: Optional[List[FilterScope]]
|
|
extra_filter: list = []
|
|
o: Optional[SortModel]
|
|
p: Optional[Page] = Page()
|
|
s: Optional[SearchModel]
|
|
is_deleted: Optional[bool]
|
|
|
|
def limit(self):
|
|
return self.p.s
|
|
|
|
def order(self):
|
|
return self.p.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 scopes(self):
|
|
return self.s
|
|
|
|
def __init__(self, **kwargs):
|
|
super().__init__(**kwargs)
|
|
_input = {**{'p': {'c': 1, 's': 5},
|
|
'f': [], 's': {}, 'o': {}}, **(kwargs or {})}
|
|
self.p = TableEngineInput.Page(**_input.get('p'))
|
|
|
|
if _input.get('f'):
|
|
self.f = [
|
|
TableEngineInput.FilterScope(
|
|
**f) for f in filter(
|
|
lambda fs:TableEngineInput.FilterScope.ValidScope(fs),
|
|
_input.get('f'))]
|
|
self.f = list(map(
|
|
lambda fs: TableEngineInput.FilterScope.CleanItems(fs),
|
|
self.f))
|
|
|
|
if _input.get('s'):
|
|
self.s = TableEngineInput.SearchModel(**_input.get('s'))
|
|
|
|
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):
|
|
return self.p.s if self.p else 20
|
|
|
|
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 self.s.cols is not None and len(
|
|
self.s.cols) > 0 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
|