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"