118 lines
3.8 KiB
Python
118 lines
3.8 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
|
|
|
|
def __new__(
|
|
cls,
|
|
message: str,
|
|
status_code: str = "HTTP_200_OK",
|
|
result: Union[Any, list] = None,
|
|
completed: bool = True,
|
|
response_model: Any = None,
|
|
):
|
|
cls.status_code = getattr(status, status_code, "HTTP_200_OK")
|
|
cls.message = message
|
|
cls.result = result
|
|
cls.completed = completed
|
|
|
|
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=None,
|
|
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=None,
|
|
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=None,
|
|
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=None,
|
|
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 = round(total_count / int(filter_model.size), 0)
|
|
|
|
pagination_dict = {
|
|
"size/total_count": [cls.result.count, total_count],
|
|
"page/total_page": [filter_model.page, total_page_number],
|
|
"order_field": filter_model.order_field,
|
|
"order_type": filter_model.order_type,
|
|
}
|
|
include_joins = dict(
|
|
include_joins=(
|
|
filter_model.include_joins if filter_model.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,
|
|
count=cls.result.count,
|
|
pagination=pagination_dict,
|
|
message=cls.message,
|
|
completed=cls.completed,
|
|
data=data,
|
|
),
|
|
)
|