|
|
"""
|
|
|
HTTP 响应包装工具
|
|
|
|
|
|
提供统一的 API 响应格式包装函数和分页工具
|
|
|
"""
|
|
|
|
|
|
from typing import Any, Optional, Union
|
|
|
from flask import request
|
|
|
from urllib.parse import urlencode
|
|
|
import math
|
|
|
|
|
|
|
|
|
# ==================== 包装函数 ====================
|
|
|
|
|
|
|
|
|
def success(data: Any = None, message: str = "成功", code: int = 200) -> dict:
|
|
|
"""
|
|
|
成功响应包装
|
|
|
|
|
|
Args:
|
|
|
data: 返回数据
|
|
|
message: 提示信息（默认 "成功"）
|
|
|
code: 业务状态码（默认 200）
|
|
|
|
|
|
Returns:
|
|
|
符合 BaseResponse 格式的字典
|
|
|
|
|
|
Example:
|
|
|
>>> return success({'id': 1, 'name': 'test'})
|
|
|
{'data': {'id': 1, 'name': 'test'}, 'code': 200, 'message': '成功'}
|
|
|
|
|
|
>>> return success([1, 2, 3], message='查询成功')
|
|
|
{'data': [1, 2, 3], 'code': 200, 'message': '查询成功'}
|
|
|
"""
|
|
|
return {"data": data, "code": code, "message": message}
|
|
|
|
|
|
|
|
|
def fail(message: str = "操作失败", code: int = 500, data: Any = None) -> dict:
|
|
|
"""
|
|
|
失败响应包装（HTTP 状态码保持 200，由前端根据 code 判断）
|
|
|
|
|
|
Args:
|
|
|
message: 错误信息
|
|
|
code: 业务错误码（默认 500）
|
|
|
data: 额外数据（如验证错误详情）
|
|
|
|
|
|
Returns:
|
|
|
符合 BaseResponse 格式的字典
|
|
|
|
|
|
Example:
|
|
|
>>> return fail('参数错误', code=400)
|
|
|
{'data': None, 'code': 400, 'message': '参数错误'}
|
|
|
|
|
|
>>> return fail('未找到资源', code=404)
|
|
|
{'data': None, 'code': 404, 'message': '未找到资源'}
|
|
|
|
|
|
>>> return fail('验证失败', code=422, data={'username': ['必填项']})
|
|
|
{'data': {'username': ['必填项']}, 'code': 422, 'message': '验证失败'}
|
|
|
"""
|
|
|
return {"data": data, "code": code, "message": message}
|
|
|
|
|
|
|
|
|
def page(
|
|
|
items: Union[list, Any],
|
|
|
pagination: Union[dict, Any, None] = None,
|
|
|
message: str = "成功",
|
|
|
code: int = 200,
|
|
|
) -> dict:
|
|
|
"""
|
|
|
分页响应包装（智能识别多种输入方式）
|
|
|
|
|
|
支持三种调用方式：
|
|
|
|
|
|
1. 传入 SQLAlchemy Pagination 对象（自动解析）
|
|
|
>>> db_pagination = User.query.paginate(page=1, per_page=10)
|
|
|
>>> return page(db_pagination)
|
|
|
|
|
|
2. 传入数据列表 + 分页信息字典
|
|
|
>>> users = [{'id': 1}, {'id': 2}]
|
|
|
>>> pagination_info = pagination_builder(None, page=1, size=10, total=100)
|
|
|
>>> return page(users, pagination_info)
|
|
|
|
|
|
3. 传入数据列表 + SQLAlchemy Pagination 对象
|
|
|
>>> db_pagination = User.query.paginate(page=1, per_page=10)
|
|
|
>>> users = UserSchema(many=True).dump(db_pagination.items)
|
|
|
>>> return page(users, db_pagination)
|
|
|
|
|
|
Args:
|
|
|
items: 数据列表 或 SQLAlchemy Pagination 对象
|
|
|
pagination: 分页信息字典 或 SQLAlchemy Pagination 对象 或 None
|
|
|
message: 提示信息
|
|
|
code: 状态码
|
|
|
|
|
|
Returns:
|
|
|
符合 BaseResponse 格式的分页数据
|
|
|
|
|
|
Example:
|
|
|
{
|
|
|
'data': {
|
|
|
'items': [...],
|
|
|
'page': {
|
|
|
'page': 1,
|
|
|
'size': 10,
|
|
|
'pages': 10,
|
|
|
'total': 100,
|
|
|
'current': '...',
|
|
|
'next': '...',
|
|
|
'prev': None,
|
|
|
'first': '...',
|
|
|
'last': '...'
|
|
|
}
|
|
|
},
|
|
|
'code': 200,
|
|
|
'message': '成功'
|
|
|
}
|
|
|
"""
|
|
|
# 方式1: items 是 SQLAlchemy Pagination 对象
|
|
|
if pagination is None and hasattr(items, "items") and hasattr(items, "total"):
|
|
|
pagination_obj = items
|
|
|
items = pagination_obj.items
|
|
|
pagination = pagination_builder(pagination_obj)
|
|
|
|
|
|
# 方式2: pagination 是 SQLAlchemy Pagination 对象
|
|
|
elif (
|
|
|
pagination is not None
|
|
|
and hasattr(pagination, "items")
|
|
|
and hasattr(pagination, "total")
|
|
|
):
|
|
|
pagination_obj = pagination
|
|
|
pagination = pagination_builder(pagination_obj)
|
|
|
|
|
|
# 方式3: pagination 已经是字典，直接使用
|
|
|
# items 已经是列表，直接使用
|
|
|
|
|
|
return {
|
|
|
"data": {"items": items, "page": pagination},
|
|
|
"code": code,
|
|
|
"message": message,
|
|
|
}
|
|
|
|
|
|
|
|
|
# ==================== 分页构建器（参考 APIFlask helpers.py）====================
|
|
|
|
|
|
|
|
|
def pagination_builder(
|
|
|
pagination: Any,
|
|
|
*,
|
|
|
page: Optional[int] = None,
|
|
|
size: Optional[int] = None,
|
|
|
total: Optional[int] = None,
|
|
|
pages: Optional[int] = None,
|
|
|
) -> dict:
|
|
|
"""
|
|
|
构建分页信息（参考 APIFlask 实现）
|
|
|
|
|
|
支持两种调用方式：
|
|
|
|
|
|
1. 传入 SQLAlchemy Pagination 对象（自动解析）
|
|
|
>>> db_pagination = User.query.paginate(page=1, per_page=10)
|
|
|
>>> pagination_info = pagination_builder(db_pagination)
|
|
|
|
|
|
2. 传入自定义参数（手动构建）
|
|
|
>>> pagination_info = pagination_builder(
|
|
|
... None,
|
|
|
... page=1,
|
|
|
... size=10,
|
|
|
... total=100
|
|
|
... )
|
|
|
|
|
|
Args:
|
|
|
pagination: SQLAlchemy Pagination 对象 或 None
|
|
|
page: 当前页码（手动模式）
|
|
|
size: 每页数量（手动模式）
|
|
|
total: 总记录数（手动模式）
|
|
|
pages: 总页数（可选，自动计算）
|
|
|
|
|
|
Returns:
|
|
|
符合 PaginationSchema 的字典
|
|
|
|
|
|
Example:
|
|
|
# 自动模式
|
|
|
>>> db_pagination = User.query.paginate(page=1, per_page=10)
|
|
|
>>> info = pagination_builder(db_pagination)
|
|
|
{'page': 1, 'size': 10, 'pages': 10, 'total': 100, ...}
|
|
|
|
|
|
# 手动模式
|
|
|
>>> info = pagination_builder(None, page=1, size=20, total=200)
|
|
|
{'page': 1, 'size': 20, 'pages': 10, 'total': 200, ...}
|
|
|
"""
|
|
|
# 自动模式：从 SQLAlchemy Pagination 对象提取
|
|
|
if pagination is not None and hasattr(pagination, "page"):
|
|
|
page = pagination.page
|
|
|
size = pagination.per_page
|
|
|
pages = pagination.pages
|
|
|
total = pagination.total
|
|
|
has_prev = pagination.has_prev
|
|
|
has_next = pagination.has_next
|
|
|
prev_num = pagination.prev_num if has_prev else None
|
|
|
next_num = pagination.next_num if has_next else None
|
|
|
|
|
|
# 手动模式：使用传入的参数
|
|
|
else:
|
|
|
page = page or 1
|
|
|
size = size or 10
|
|
|
total = total or 0
|
|
|
|
|
|
# 自动计算总页数
|
|
|
if pages is None:
|
|
|
pages = math.ceil(total / size) if size > 0 else 0
|
|
|
|
|
|
# 计算上下页
|
|
|
has_prev = page > 1
|
|
|
has_next = page < pages
|
|
|
prev_num = page - 1 if has_prev else None
|
|
|
next_num = page + 1 if has_next else None
|
|
|
|
|
|
# 生成分页 URL（参考 APIFlask 实现）
|
|
|
current_url = _generate_page_url(page, size)
|
|
|
next_url = _generate_page_url(next_num, size) if has_next else None
|
|
|
prev_url = _generate_page_url(prev_num, size) if has_prev else None
|
|
|
first_url = _generate_page_url(1, size)
|
|
|
last_url = _generate_page_url(pages, size) if pages > 0 else None
|
|
|
|
|
|
return {
|
|
|
"page": page,
|
|
|
"size": size, # ✅ 使用 size 代替 per_page
|
|
|
"pages": pages,
|
|
|
"total": total,
|
|
|
"current": current_url,
|
|
|
"next": next_url,
|
|
|
"prev": prev_url,
|
|
|
"first": first_url,
|
|
|
"last": last_url,
|
|
|
}
|
|
|
|
|
|
|
|
|
def _generate_page_url(page_num: Optional[int], page_size: int) -> Optional[str]:
|
|
|
"""
|
|
|
生成分页 URL（内部辅助函数）
|
|
|
|
|
|
参考 APIFlask 的 URL 生成逻辑
|
|
|
|
|
|
Args:
|
|
|
page_num: 页码
|
|
|
page_size: 每页数量
|
|
|
|
|
|
Returns:
|
|
|
完整的分页 URL 或 None
|
|
|
"""
|
|
|
if page_num is None:
|
|
|
return None
|
|
|
|
|
|
try:
|
|
|
# 获取当前请求的 URL 和查询参数
|
|
|
base_url = request.base_url
|
|
|
args = request.args.copy()
|
|
|
|
|
|
# 更新分页参数
|
|
|
args["page"] = page_num
|
|
|
args["size"] = page_size # ✅ 使用 size 参数名
|
|
|
|
|
|
# 构建完整 URL
|
|
|
if args:
|
|
|
return f"{base_url}?{urlencode(args)}"
|
|
|
return base_url
|
|
|
except RuntimeError:
|
|
|
# 在请求上下文之外调用时返回 None
|
|
|
return None
|