178 lines
4.5 KiB
Python
178 lines
4.5 KiB
Python
"""
|
|
统一的API响应工具
|
|
提供标准化的API响应格式
|
|
"""
|
|
|
|
from flask import jsonify, Response
|
|
from typing import Any, Optional, Dict
|
|
from werkzeug.http import HTTP_STATUS_CODES
|
|
|
|
|
|
class APIResponse:
|
|
"""API响应类"""
|
|
|
|
@staticmethod
|
|
def success(data: Any = None, message: str = "操作成功", code: int = 200) -> Response:
|
|
"""
|
|
成功响应
|
|
|
|
Args:
|
|
data: 响应数据
|
|
message: 响应消息
|
|
code: HTTP状态码
|
|
|
|
Returns:
|
|
Response: Flask响应对象
|
|
"""
|
|
response_data = {
|
|
'success': True,
|
|
'message': message
|
|
}
|
|
if data is not None:
|
|
response_data['data'] = data
|
|
return jsonify(response_data), code
|
|
|
|
@staticmethod
|
|
def error(message: str = "操作失败", code: int = 400, data: Any = None) -> Response:
|
|
"""
|
|
错误响应
|
|
|
|
Args:
|
|
message: 错误消息
|
|
code: HTTP状态码
|
|
data: 额外数据
|
|
|
|
Returns:
|
|
Response: Flask响应对象
|
|
"""
|
|
response_data = {
|
|
'success': False,
|
|
'message': message
|
|
}
|
|
if data is not None:
|
|
response_data['data'] = data
|
|
return jsonify(response_data), code
|
|
|
|
@staticmethod
|
|
def validation_error(errors: Dict[str, str], message: str = "参数验证失败") -> Response:
|
|
"""
|
|
参数验证错误响应
|
|
|
|
Args:
|
|
errors: 错误详情字典
|
|
message: 错误消息
|
|
|
|
Returns:
|
|
Response: Flask响应对象
|
|
"""
|
|
return APIResponse.error(
|
|
message=message,
|
|
code=400,
|
|
data={'errors': errors}
|
|
)
|
|
|
|
@staticmethod
|
|
def unauthorized(message: str = "未授权，请先登录") -> Response:
|
|
"""
|
|
未授权响应
|
|
|
|
Args:
|
|
message: 错误消息
|
|
|
|
Returns:
|
|
Response: Flask响应对象
|
|
"""
|
|
return APIResponse.error(message=message, code=401)
|
|
|
|
@staticmethod
|
|
def forbidden(message: str = "权限不足") -> Response:
|
|
"""
|
|
禁止访问响应
|
|
|
|
Args:
|
|
message: 错误消息
|
|
|
|
Returns:
|
|
Response: Flask响应对象
|
|
"""
|
|
return APIResponse.error(message=message, code=403)
|
|
|
|
@staticmethod
|
|
def not_found(message: str = "资源不存在") -> Response:
|
|
"""
|
|
资源不存在响应
|
|
|
|
Args:
|
|
message: 错误消息
|
|
|
|
Returns:
|
|
Response: Flask响应对象
|
|
"""
|
|
return APIResponse.error(message=message, code=404)
|
|
|
|
@staticmethod
|
|
def server_error(message: str = "服务器内部错误") -> Response:
|
|
"""
|
|
服务器错误响应
|
|
|
|
Args:
|
|
message: 错误消息
|
|
|
|
Returns:
|
|
Response: Flask响应对象
|
|
"""
|
|
return APIResponse.error(message=message, code=500)
|
|
|
|
@staticmethod
|
|
def paginated(items: list, total: int, page: int, per_page: int,
|
|
message: str = "查询成功") -> Response:
|
|
"""
|
|
分页响应
|
|
|
|
Args:
|
|
items: 数据列表
|
|
total: 总数
|
|
page: 当前页
|
|
per_page: 每页数量
|
|
message: 响应消息
|
|
|
|
Returns:
|
|
Response: Flask响应对象
|
|
"""
|
|
import math
|
|
total_pages = math.ceil(total / per_page) if total > 0 else 0
|
|
|
|
return APIResponse.success(
|
|
data={
|
|
'items': items,
|
|
'pagination': {
|
|
'current_page': page,
|
|
'per_page': per_page,
|
|
'total': total,
|
|
'total_pages': total_pages,
|
|
'has_prev': page > 1,
|
|
'has_next': page < total_pages
|
|
}
|
|
},
|
|
message=message
|
|
)
|
|
|
|
|
|
def success_response(data: Any = None, message: str = "操作成功") -> Response:
|
|
"""成功响应的便捷函数"""
|
|
return APIResponse.success(data=data, message=message)
|
|
|
|
|
|
def error_response(message: str = "操作失败", code: int = 400) -> Response:
|
|
"""错误响应的便捷函数"""
|
|
return APIResponse.error(message=message, code=code)
|
|
|
|
|
|
def not_found_response(message: str = "资源不存在") -> Response:
|
|
"""资源不存在响应的便捷函数"""
|
|
return APIResponse.not_found(message=message)
|
|
|
|
|
|
def validation_error_response(errors: Dict[str, str]) -> Response:
|
|
"""参数验证错误响应的便捷函数"""
|
|
return APIResponse.validation_error(errors=errors) |