129 lines
4.0 KiB
Python
129 lines
4.0 KiB
Python
from typing import Any, Dict, Generic, Optional, TypeVar, Union
|
||
from pydantic import BaseModel, Field
|
||
from datetime import datetime
|
||
import uuid
|
||
from fastapi.responses import JSONResponse
|
||
from fastapi import status
|
||
|
||
# 定义泛型类型变量
|
||
T = TypeVar('T')
|
||
|
||
class ResponseModel(BaseModel, Generic[T]):
|
||
"""统一API响应模型"""
|
||
code: int = Field(0, description="状态码，0表示成功，非0表示错误")
|
||
message: str = Field("success", description="响应消息")
|
||
data: Optional[T] = Field(None, description="响应数据")
|
||
request_id: str = Field(default_factory=lambda: str(uuid.uuid4())[:8], description="请求ID")
|
||
timestamp: datetime = Field(default_factory=datetime.now, description="响应时间戳")
|
||
|
||
class Config:
|
||
arbitrary_types_allowed = True
|
||
json_encoders = {
|
||
datetime: lambda dt: dt.isoformat()
|
||
}
|
||
|
||
class ResponseFactory:
|
||
"""响应工厂类，用于生成标准格式的响应"""
|
||
|
||
@staticmethod
|
||
def success(data: Any = None, message: str = "success") -> Dict[str, Any]:
|
||
"""生成成功响应"""
|
||
return ResponseModel(
|
||
code=200,
|
||
message=message,
|
||
data=data
|
||
).model_dump(exclude_none=True)
|
||
|
||
@staticmethod
|
||
def error(code: int = 1, message: str = "error", data: Any = None) -> Dict[str, Any]:
|
||
"""生成错误响应"""
|
||
return ResponseModel(
|
||
code=code,
|
||
message=message,
|
||
data=data
|
||
).model_dump(exclude_none=True)
|
||
|
||
@staticmethod
|
||
def failed_validation(message: str = "数据验证失败", errors: Any = None) -> Dict[str, Any]:
|
||
"""生成验证失败响应"""
|
||
return ResponseModel(
|
||
code=400,
|
||
message=message,
|
||
data=errors
|
||
).model_dump(exclude_none=True)
|
||
|
||
@staticmethod
|
||
def not_found(message: str = "资源不存在", data: Any = None) -> Dict[str, Any]:
|
||
"""生成资源不存在响应"""
|
||
return ResponseModel(
|
||
code=404,
|
||
message=message,
|
||
data=data
|
||
).model_dump(exclude_none=True)
|
||
|
||
@staticmethod
|
||
def unauthorized(message: str = "未授权访问", data: Any = None) -> Dict[str, Any]:
|
||
"""生成未授权响应"""
|
||
return ResponseModel(
|
||
code=401,
|
||
message=message,
|
||
data=data
|
||
).model_dump(exclude_none=True)
|
||
|
||
@staticmethod
|
||
def forbidden(message: str = "禁止访问", data: Any = None) -> Dict[str, Any]:
|
||
"""生成禁止访问响应"""
|
||
return ResponseModel(
|
||
code=403,
|
||
message=message,
|
||
data=data
|
||
).model_dump(exclude_none=True)
|
||
|
||
@staticmethod
|
||
def server_error(message: str = "服务器内部错误", data: Any = None) -> Dict[str, Any]:
|
||
"""生成服务器错误响应"""
|
||
return ResponseModel(
|
||
code=500,
|
||
message=message,
|
||
data=data
|
||
).model_dump(exclude_none=True)
|
||
|
||
@staticmethod
|
||
def json_response(
|
||
data: Any = None,
|
||
message: str = "success",
|
||
code: int = 0,
|
||
status_code: int = status.HTTP_200_OK
|
||
) -> JSONResponse:
|
||
"""生成JSONResponse对象"""
|
||
content = ResponseModel(
|
||
code=code,
|
||
message=message,
|
||
data=data
|
||
).model_dump(exclude_none=True)
|
||
|
||
return JSONResponse(
|
||
content=content,
|
||
status_code=status_code
|
||
)
|
||
|
||
class Result:
|
||
"""统一API响应工具类"""
|
||
|
||
@staticmethod
|
||
def success(data: Any = None, message: str = "success") -> Dict[str, Any]:
|
||
"""返回成功响应"""
|
||
return {
|
||
"code": 0,
|
||
"message": message,
|
||
"data": data
|
||
}
|
||
|
||
@staticmethod
|
||
def error(code: int = 1, message: str = "error", data: Any = None) -> Dict[str, Any]:
|
||
"""返回错误响应"""
|
||
return {
|
||
"code": code,
|
||
"message": message,
|
||
"data": data
|
||
} |