sakhteman/core/utils/te/utils/Models.py

from enum import Enum
from typing import Any
from typing import Generic, TypeVar, List, Optional

import pydantic
from pydantic import BaseModel
from pydantic.generics import GenericModel


class FilterItem(BaseModel):
    col: Optional[str]
    cr: Optional[str]
    id: Optional[str]
    op: Optional[str]
    type: Optional[str]
    val: Optional[Any]
    hidden: Optional[bool]

    @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 OutPutType(str, Enum):
    JSON = 'json'
    EXEL = 'exel'
    PDF = 'pdf'


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
    force: Optional[bool] = False


class SearchModel(BaseModel):
    text: Optional[str]
    cols: Optional[List[Any]]


class ExportModel(BaseModel):
    ot: Optional[OutPutType]
    c: Optional[dict]
    d: Optional[str]


class SortModel(BaseModel):
    field: Optional[str]
    direction: Optional[str]


class MultiSortModel(BaseModel):
    field: Optional[str]
    direction: Optional[str]
    fields: Optional[List[SortModel]]


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[MultiSortModel]
    p: Optional[Page] = Page()
    s: Optional[SearchModel]
    e: Optional[ExportModel]
    is_deleted: Optional[bool]

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        _input = {**{'p': {'c': 1, 's': 5},
                     'f': [], 's': {}, 'o': {}, 'e': {}}, **(kwargs or {})}
        self.p = Page(**_input.get('p'))
        self.o = MultiSortModel(**_input.get('o'))
        self.e = ExportModel(**_input.get('e'))
        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):
        # HIT  : LIMIT 0 : in Mongo Return all records
        limit = self.p.s if self.p else 20
        if self.p.force:
            return 1000000
        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"


T = TypeVar('T')


class TableEngineInputType(GenericModel, Generic[T]):
    f: Optional[List]
    o: Optional[dict] = {}
    p: Optional[dict] = {}
    s: Optional[dict] = {}
    e: Optional[dict] = {}


class TableEngineExport(GenericModel, Generic[T]):
    rows: Optional[List[T]]
    total: Optional[int]
    page: Optional[int]
    te: Optional[TableEngineInputType]


class TableEngineRequestType(GenericModel, Generic[T]):
    te: Optional[TableEngineInputType]
    extra: Optional[dict]