""" 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