142 lines
4.3 KiB
Python
142 lines
4.3 KiB
Python
from typing import Any, Dict, List, Optional, Union, ClassVar
|
||
from fastapi import status
|
||
from fastapi.responses import JSONResponse
|
||
import os
|
||
|
||
from app.models.api_response import APIResponseModel
|
||
|
||
class APIResponse:
|
||
"""API标准响应格式工具类"""
|
||
|
||
# 全局开关，控制是否使用标准响应格式
|
||
USE_STANDARD_RESPONSE = os.environ.get("USE_STANDARD_RESPONSE", "1") == "1"
|
||
|
||
@staticmethod
|
||
def ok(data: Any = None, message: str = "操作成功") -> Dict[str, Any]:
|
||
"""成功响应"""
|
||
if not APIResponse.USE_STANDARD_RESPONSE:
|
||
return data
|
||
|
||
return APIResponseModel(
|
||
code=200,
|
||
message=message,
|
||
data=data
|
||
).dict()
|
||
|
||
@staticmethod
|
||
def error(message: str = "操作失败", code: int = 400, data: Any = None) -> Dict[str, Any]:
|
||
"""错误响应"""
|
||
# 错误响应即使禁用了标准格式也返回标准格式，确保错误信息清晰
|
||
return APIResponseModel(
|
||
code=code,
|
||
message=message,
|
||
data=data
|
||
).dict()
|
||
|
||
@staticmethod
|
||
def format_response(data: Any, message: str = "操作成功", code: int = 200) -> Dict:
|
||
"""
|
||
根据环境变量决定是否返回标准格式的响应
|
||
|
||
如果USE_STANDARD_RESPONSE=1，返回标准格式：
|
||
{
|
||
"code": 200,
|
||
"message": "操作成功",
|
||
"data": {...}
|
||
}
|
||
|
||
如果USE_STANDARD_RESPONSE=0，直接返回data
|
||
"""
|
||
if APIResponse.USE_STANDARD_RESPONSE:
|
||
return APIResponseModel(code=code, message=message, data=data).dict()
|
||
else:
|
||
return data if data is not None else {"message": message}
|
||
|
||
@staticmethod
|
||
def json_response(
|
||
data: Any = None,
|
||
message: str = "操作成功",
|
||
code: int = 200,
|
||
success: bool = True,
|
||
status_code: int = status.HTTP_200_OK,
|
||
headers: Dict[str, str] = None
|
||
) -> JSONResponse:
|
||
"""
|
||
返回JSONResponse对象
|
||
|
||
Args:
|
||
data: 响应数据
|
||
message: 响应消息
|
||
code: 业务状态码
|
||
success: 是否成功
|
||
status_code: HTTP状态码
|
||
headers: 自定义响应头
|
||
|
||
Returns:
|
||
JSONResponse对象
|
||
"""
|
||
if not APIResponse.USE_STANDARD_RESPONSE and success:
|
||
return JSONResponse(
|
||
content=data,
|
||
status_code=status_code,
|
||
headers=headers
|
||
)
|
||
|
||
content = APIResponseModel(
|
||
code=code,
|
||
message=message,
|
||
data=data
|
||
).dict()
|
||
|
||
return JSONResponse(
|
||
content=content,
|
||
status_code=status_code,
|
||
headers=headers
|
||
)
|
||
|
||
# 常用响应码封装
|
||
@classmethod
|
||
def created(cls, data: Any = None, message: str = "创建成功") -> Dict[str, Any]:
|
||
"""201 创建成功"""
|
||
if not cls.USE_STANDARD_RESPONSE:
|
||
return data
|
||
return cls.ok(data, message)
|
||
|
||
@classmethod
|
||
def accepted(cls, data: Any = None, message: str = "请求已接受") -> Dict[str, Any]:
|
||
"""202 已接受"""
|
||
if not cls.USE_STANDARD_RESPONSE:
|
||
return data
|
||
return cls.ok(data, message)
|
||
|
||
@classmethod
|
||
def no_content(cls) -> Dict[str, Any]:
|
||
"""204 无内容"""
|
||
if not cls.USE_STANDARD_RESPONSE:
|
||
return None
|
||
return cls.ok(None, "无内容")
|
||
|
||
@classmethod
|
||
def bad_request(cls, message: str = "请求参数错误") -> Dict[str, Any]:
|
||
"""400 请求错误"""
|
||
return cls.error(message, 400)
|
||
|
||
@classmethod
|
||
def unauthorized(cls, message: str = "未授权") -> Dict[str, Any]:
|
||
"""401 未授权"""
|
||
return cls.error(message, 401)
|
||
|
||
@classmethod
|
||
def forbidden(cls, message: str = "禁止访问") -> Dict[str, Any]:
|
||
"""403 禁止"""
|
||
return cls.error(message, 403)
|
||
|
||
@classmethod
|
||
def not_found(cls, message: str = "资源不存在") -> Dict[str, Any]:
|
||
"""404 不存在"""
|
||
return cls.error(message, 404)
|
||
|
||
@classmethod
|
||
def server_error(cls, message: str = "服务器内部错误") -> Dict[str, Any]:
|
||
"""500 服务器错误"""
|
||
return cls.error(message, 500) |