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