277 lines
8.3 KiB
Python
277 lines
8.3 KiB
Python
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
|
|
from api_objects.errors.errorHandlers import HTTPExceptionEvyos
|
|
|
|
|
|
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: int = 1,
|
|
size: int = 10,
|
|
order_field: str = "id",
|
|
order_type: str = "asc",
|
|
):
|
|
self.size = size if 10 < size < 40 else 10
|
|
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()
|
|
if self.page > self.totalPage:
|
|
self.page = self.totalPage
|
|
elif self.page < 1:
|
|
self.page = 1
|
|
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()
|
|
self.setter_page()
|
|
|
|
def setter_page(self):
|
|
self.pageCount = self.size
|
|
self.totalPage = int(round(self.totalCount / self.size, 0))
|
|
remainder = self.totalCount % self.size
|
|
if remainder > 0:
|
|
self.totalPage = int(round(self.totalCount / self.size, 0)) + 1
|
|
if self.page == self.totalPage:
|
|
self.pageCount = remainder
|
|
|
|
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: Any
|
|
message: str
|
|
completed: bool
|
|
cls_object: Any = (None,)
|
|
|
|
def __new__(
|
|
cls,
|
|
message: str,
|
|
response_model: Any,
|
|
status_code: str = "HTTP_200_OK",
|
|
result: AlchemyResponse = None,
|
|
completed: bool = True,
|
|
cls_object: 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
|
|
|
|
if not isinstance(cls.result, AlchemyResponse):
|
|
HTTPExceptionEvyos(
|
|
lang=cls_object.lang,
|
|
error_code="HTTP_400_BAD_REQUEST",
|
|
)
|
|
|
|
if not cls.result.first:
|
|
HTTPExceptionEvyos(
|
|
lang=cls_object.lang,
|
|
error_code="HTTP_400_BAD_REQUEST",
|
|
)
|
|
|
|
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: Any = 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: Any = 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:
|
|
HTTPExceptionEvyos(
|
|
lang=cls_object.lang,
|
|
error_code="HTTP_400_BAD_REQUEST",
|
|
)
|
|
|
|
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: Any = (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: Any = 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: Any = response_model
|
|
|
|
if not isinstance(cls.result, list):
|
|
HTTPExceptionEvyos(
|
|
lang=cls_object.lang,
|
|
error_code="HTTP_400_BAD_REQUEST",
|
|
)
|
|
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: Any = (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: Any = 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: Any = response_model
|
|
|
|
if not isinstance(cls.result, dict):
|
|
HTTPExceptionEvyos(
|
|
lang=cls_object.lang,
|
|
error_code="HTTP_400_BAD_REQUEST",
|
|
)
|
|
|
|
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,
|
|
),
|
|
)
|