148 lines
5.2 KiB
Python
148 lines
5.2 KiB
Python
from typing import Any, Union
|
|
from fastapi import status
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from databases.sql_models.response_model import AlchemyResponse
|
|
|
|
|
|
class AlchemyJsonResponse:
|
|
status_code: status
|
|
message: str
|
|
result: AlchemyResponse
|
|
completed: bool
|
|
filter_attributes: Any = None
|
|
response_model: Any = None
|
|
cls_object: Any = None
|
|
|
|
@staticmethod
|
|
def get_total_count(cls_object, filter_attributes):
|
|
total_page_number = 1
|
|
count_to_use = cls_object.total_count / int(filter_attributes.size)
|
|
if cls_object.total_count > int(filter_attributes.size):
|
|
if isinstance(count_to_use, int):
|
|
total_page_number = round(count_to_use, 0)
|
|
elif isinstance(count_to_use, float):
|
|
total_page_number = round(count_to_use, 0) + 1
|
|
return total_page_number
|
|
|
|
def __new__(
|
|
cls,
|
|
message: str,
|
|
status_code: str = "HTTP_200_OK",
|
|
result: Union[Any, list] = 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
|
|
|
|
pagination_dict = {
|
|
"size/total_count": [10, 10],
|
|
"page/total_page": [1, 1],
|
|
"order_field": "id",
|
|
"order_type": "asc",
|
|
}
|
|
if filter_attributes:
|
|
total_page_number = cls.get_total_count(cls_object, filter_attributes)
|
|
pagination_dict = {
|
|
"size/total_count": [filter_attributes.size, cls_object.total_count],
|
|
"page/total_page": [filter_attributes.page, total_page_number],
|
|
"order_field": filter_attributes.order_field,
|
|
"order_type": filter_attributes.order_type,
|
|
}
|
|
|
|
if isinstance(cls.result, dict) or isinstance(cls.result, list):
|
|
return JSONResponse(
|
|
status_code=cls.status_code,
|
|
content=dict(
|
|
total_count=len(cls.result),
|
|
count=len(cls.result),
|
|
pagination=pagination_dict,
|
|
completed=cls.completed,
|
|
message=cls.message,
|
|
data=cls.result,
|
|
),
|
|
)
|
|
|
|
first_item = getattr(cls.result, "data", None)
|
|
if not first_item:
|
|
return JSONResponse(
|
|
status_code=cls.status_code,
|
|
content=dict(
|
|
total_count=0,
|
|
count=0,
|
|
pagination=pagination_dict,
|
|
completed=cls.completed,
|
|
message=cls.message,
|
|
data=[],
|
|
),
|
|
)
|
|
|
|
if cls.result.first:
|
|
return JSONResponse(
|
|
status_code=cls.status_code,
|
|
content=dict(
|
|
total_count=1,
|
|
count=1,
|
|
pagination=pagination_dict,
|
|
completed=cls.completed,
|
|
message=cls.message,
|
|
data=cls.result.data.get_dict(),
|
|
),
|
|
)
|
|
|
|
if not cls.result.get(1).filter_attr and isinstance(cls.result.data, list):
|
|
counts = cls.result.count
|
|
return JSONResponse(
|
|
status_code=cls.status_code,
|
|
content=dict(
|
|
total_count=counts,
|
|
count=counts,
|
|
pagination=pagination_dict,
|
|
completed=cls.completed,
|
|
message=cls.message,
|
|
data=[result_data.get_dict() for result_data in cls.result.data],
|
|
),
|
|
)
|
|
|
|
# filter_model = cls.result.get(1).filter_attr
|
|
total_count = cls.result.get(1).query.limit(None).offset(None).count()
|
|
total_page_number = cls.get_total_count(cls_object, filter_attributes)
|
|
pagination_dict = {
|
|
"size/total_count": [filter_attributes.size, cls_object.total_count],
|
|
"page/total_page": [filter_attributes.page, total_page_number],
|
|
"order_field": filter_attributes.order_field,
|
|
"order_type": filter_attributes.order_type,
|
|
}
|
|
include_joins = dict(
|
|
include_joins=(
|
|
filter_attributes.include_joins
|
|
if filter_attributes.include_joins
|
|
else []
|
|
)
|
|
)
|
|
data = []
|
|
for data_object in cls.result.data:
|
|
data_dict = data_object.get_dict(include_joins=include_joins)
|
|
if cls.response_model:
|
|
data_dict = cls.response_model(
|
|
**data_object.get_dict(include_joins=include_joins)
|
|
).dump()
|
|
data.append(data_dict)
|
|
return JSONResponse(
|
|
status_code=cls.status_code,
|
|
content=dict(
|
|
total_count=total_count or 1,
|
|
count=cls.result.count,
|
|
pagination=pagination_dict,
|
|
message=cls.message,
|
|
completed=cls.completed,
|
|
data=data,
|
|
),
|
|
)
|