56 lines
1.6 KiB
Python
56 lines
1.6 KiB
Python
from typing import Any, Optional
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel, ConfigDict
|
|
from datetime import datetime, time, date
|
|
from typing import TypeVar, Generic
|
|
from fastapi import status
|
|
import json
|
|
from decimal import Decimal
|
|
from app.core.utils import CommonUtils
|
|
|
|
class CustomJSONEncoder(json.JSONEncoder):
|
|
def default(self, obj):
|
|
if isinstance(obj, datetime):
|
|
return CommonUtils.get_asia_datetime(obj)
|
|
if isinstance(obj, time):
|
|
return obj.strftime('%H:%M:%S')
|
|
elif isinstance(obj, date):
|
|
return obj.strftime('%Y-%m-%d')
|
|
elif isinstance(obj, Decimal):
|
|
return float(obj)
|
|
return super().default(obj)
|
|
|
|
class CustomJSONResponse(JSONResponse):
|
|
def render(self, content) -> bytes:
|
|
return json.dumps(
|
|
content,
|
|
ensure_ascii=False,
|
|
allow_nan=False,
|
|
indent=None,
|
|
separators=(",", ":"),
|
|
cls=CustomJSONEncoder
|
|
).encode("utf-8")
|
|
|
|
|
|
class ResponseModel(BaseModel):
|
|
code: int = status.HTTP_200_OK
|
|
message: str = "success"
|
|
data: Optional[Any] = None
|
|
|
|
def json_response(self):
|
|
return CustomJSONResponse(content=self.model_dump())
|
|
|
|
|
|
def success_response(*, data: Any = None, message: str = "success") -> CustomJSONResponse:
|
|
return ResponseModel(
|
|
code=status.HTTP_200_OK,
|
|
message=message,
|
|
data=data
|
|
).json_response()
|
|
|
|
def error_response(*, code: int = 400, message: str, data: Any = None) -> CustomJSONResponse:
|
|
return ResponseModel(
|
|
code=code,
|
|
message=message,
|
|
data=data
|
|
).json_response() |