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)