VberAdminPy/VapAdmin/BACKEND/utils/response_util.py

from typing import Any, Optional

from fastapi import status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse, Response, StreamingResponse

from core.constant import HttpStatusConstant


class ResponseUtil:
    """
    响应工具类
    """

    @classmethod
    def response(
        cls,
        success: bool = True,
        code: int = HttpStatusConstant.SUCCESS,
        msg: str = "操作成功",
        data: Optional[Any] = None,
    ):
        result = {
            "success": success,
            "code": code,
            "message": msg,
            "data": data,
        }

        return JSONResponse(
            status_code=status.HTTP_200_OK, content=jsonable_encoder(result)
        )

    @classmethod
    def success(cls, data: Optional[Any] = None, msg: str = None) -> Response:
        return cls.response(msg=msg if msg else "操作成功", data=data)

    @classmethod
    def failure(
        cls,
        msg: str = None,
        data: Optional[Any] = None,
        code: int = HttpStatusConstant.WARN,
    ) -> Response:
        return cls.response(False, code, msg if msg else "操作失败", data)

    @classmethod
    def unauthorized(
        cls,
        msg: str = None,
        data: Optional[Any] = None,
    ) -> Response:
        return cls.failure(
            msg if msg else "登录信息已过期，访问系统资源失败",
            data,
            HttpStatusConstant.UNAUTHORIZED,
        )

    @classmethod
    def forbidden(
        cls,
        msg: str = None,
        data: Optional[Any] = None,
    ) -> Response:
        return cls.failure(
            msg if msg else "没有权限，访问系统资源失败",
            data,
            HttpStatusConstant.FORBIDDEN,
        )

    @classmethod
    def error(
        cls,
        msg: str = None,
        data: Optional[Any] = None,
    ) -> Response:
        return cls.failure(
            msg if msg else "系统服务异常", data, HttpStatusConstant.ERROR
        )

    @classmethod
    def not_found(cls, path: Optional[str] = None) -> Response:
        return cls.failure(
            f"[{path}]资源不存在" if path else "资源不存在",
            code=HttpStatusConstant.NOT_FOUND,
        )

    @classmethod
    def streaming(cls, *, data: Any = None):
        """
        流式响应方法

        :param data: 流式传输的内容
        :return: 流式响应结果
        """
        return StreamingResponse(status_code=status.HTTP_200_OK, content=data)

    @classmethod
    def json(cls, data: Any = None):
        """
        JSON响应方法

        :param data: JSON响应的内容
        :return: JSON响应结果
        """
        return JSONResponse(status_code=status.HTTP_200_OK, content=data)