wag-managment-api-service-version-3/ApiServices/api_handlers/core_response.py

from typing import Any, Union, Callable, Any
from fastapi import status
from fastapi.responses import JSONResponse

from pydantic import BaseModel
from api_validations.validations_request import PydanticBaseModel, BaseModelRegular
from databases.sql_models.response_model import AlchemyResponse
from sqlalchemy.orm import Query

MODEL_TYPE = Callable[[Any], Any]

class Pagination:
    size: int = 10
    page: int = 1
    orderField: str = "id"
    orderType: str = "asc"
    pageCount: int = 1
    totalCount: int = 1
    totalPage: int = 1

    def change(self, page=None, size=None, order_field=None, order_type=None):
        self.page = page or self.page
        self.size = size or self.size
        self.orderField = order_field or self.orderField
        self.orderType = order_type or self.orderType
        self.setter_page()

    def feed(self, data):
        if isinstance(data, list):
            self.totalCount = len(data)
        elif isinstance(data, AlchemyResponse):
            self.totalCount = data.count
        elif isinstance(data, Query):
            self.totalCount = data.count()

    def setter_page(self):
        self.pageCount = self.size
        self.totalPage = int(round(self.totalCount / self.size, 0))
        if self.totalCount % self.size > 0:
            if self.page == self.totalPage:
                self.pageCount = self.totalCount % self.size
            self.totalPage = int(round(self.totalCount / self.size, 0)) + 1

    def as_dict(self):
        return {
            "size": self.size,
            "page": self.page,
            "totalCount": self.totalCount,
            "totalPage": self.totalPage,
            "pageCount": self.pageCount,
            "orderField": self.orderField,
            "orderType": self.orderType,
        }


class SingleAlchemyResponse:
    status_code = "HTTP_200_OK"
    result: AlchemyResponse
    response_model: MODEL_TYPE
    message: str
    completed: bool

    def __new__(
        cls,
        message: str,
        response_model: MODEL_TYPE,
        status_code: str = "HTTP_200_OK",
        result: AlchemyResponse = None,
        completed: bool = True,
    ):
        cls.status_code = getattr(status, status_code, "HTTP_200_OK")
        cls.message = message
        cls.result = result
        cls.completed = completed
        cls.response_model = response_model

        if not isinstance(cls.result, AlchemyResponse):
            raise Exception("Invalid response type 4 single alchemy response")

        if not cls.result.first:
            raise Exception("Invalid data type 4 single alchemy response")

        pagination = Pagination()
        pagination.change(page=1)
        BaseModelRegular(**cls.result.data.get_dict())
        data = cls.result.data.get_dict()
        if cls.response_model:
            data = cls.response_model(**cls.result.data.get_dict()).dump()
        return JSONResponse(
            status_code=cls.status_code,
            content=dict(
                pagination=pagination.as_dict(),
                completed=cls.completed,
                message=cls.message,
                data=data,
            ),
        )


class AlchemyJsonResponse:
    status_code: status
    message: str
    result: AlchemyResponse
    completed: bool
    filter_attributes: Any = None
    response_model: MODEL_TYPE = None
    cls_object: Any = None

    def __new__(
        cls,
        message: str,
        status_code: str = "HTTP_200_OK",
        result: Union[BaseModelRegular, BaseModel, PydanticBaseModel] = None,
        completed: bool = True,
        response_model: MODEL_TYPE = None,
        cls_object: Any = None,
        filter_attributes: Any = None,
    ):
        cls.status_code = getattr(status, status_code, "HTTP_200_OK")
        cls.message = message
        cls.result = result
        cls.completed = completed
        cls.response_model = response_model
        cls.filter_attributes = filter_attributes
        cls.cls_object = cls_object
        pagination = Pagination()

        if cls.result.first:
            raise Exception("Invalid data type 4 alchemy response")

        if filter_attributes:
            pagination.change(
                page=filter_attributes.page, size=filter_attributes.size,
                order_field=filter_attributes.order_field,
                order_type=filter_attributes.order_type,
            )
        data = []
        for data_object in cls.result.data:
            data_dict = data_object.get_dict()
            if cls.response_model:
                data_dict = cls.response_model(**data_object.get_dict()).dump()
            data.append(data_dict)
        pagination.feed(data)
        return JSONResponse(
            status_code=cls.status_code,
            content=dict(
                pagination=pagination.as_dict(),
                message=cls.message,
                completed=cls.completed,
                data=data,
            ),
        )


class ListJsonResponse:
    status_code = "HTTP_200_OK"
    result: list
    message: str
    completed: bool
    filter_attributes: Any
    response_model: MODEL_TYPE = None,
    cls_object: Any = None,

    def __new__(
        cls,
        message: str,
        status_code: str = "HTTP_200_OK",
        result: Union[BaseModelRegular, BaseModel, PydanticBaseModel] = None,
        completed: bool = True,
        response_model: MODEL_TYPE = None,
        cls_object: Any = None,
        filter_attributes: Any = None,
    ):
        cls.status_code = getattr(status, status_code, "HTTP_200_OK")
        cls.message = message
        cls.result = result
        cls.completed = completed
        cls.filter_attributes = filter_attributes
        cls.response_model: MODEL_TYPE = response_model

        if not isinstance(cls.result, list):
            raise Exception("Invalid data type 4 list json response")
        pagination = Pagination()
        pagination.change(page=1)
        data = list(cls.result)
        if cls.response_model:
            data = [cls.response_model(**data_object).dump() for data_object in cls.result]
        pagination.feed(data)
        return JSONResponse(
            status_code=cls.status_code,
            content=dict(
                pagination=pagination.as_dict(),
                completed=cls.completed,
                message=cls.message,
                data=cls.result,
            ),
        )

class DictJsonResponse:
    status_code = "HTTP_200_OK"
    result: dict
    message: str
    completed: bool
    filter_attributes: Any
    response_model: MODEL_TYPE = None,
    cls_object: Any = None,

    def __new__(
        cls,
        message: str,
        status_code: str = "HTTP_200_OK",
        result: Union[BaseModelRegular, BaseModel, PydanticBaseModel] = None,
        completed: bool = True,
        response_model: MODEL_TYPE = None,
        cls_object: Any = None,
        filter_attributes: Any = None,
    ):
        cls.status_code = getattr(status, status_code, "HTTP_200_OK")
        cls.message = message
        cls.result = result
        cls.completed = completed
        cls.filter_attributes = filter_attributes
        cls.response_model: MODEL_TYPE = response_model

        if not isinstance(cls.result, dict):
            raise Exception("Invalid data type 4 dict json response")

        pagination = Pagination()
        pagination.change(page=1)
        data = cls.result
        if cls.response_model:
            data = cls.response_model(**cls.result).dump()
        return JSONResponse(
            status_code=cls.status_code,
            content=dict(
                pagination=pagination.as_dict(),
                completed=cls.completed,
                message=cls.message,
                data=data,
            ),
        )