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 }