iTi-Flask/docs/HTTP_RESPONSE_UTILS.md

# HTTP 响应包装工具使用指南

## 📖 概述

本工具库提供统一的 API 响应格式包装函数和分页工具，基于 APIFlask 框架设计，简化 API 开发流程。

**核心特性：**
- ✅ 统一返回格式：`{data, code, message}`
- ✅ 智能分页支持：自动识别 SQLAlchemy Pagination 对象
- ✅ 灵活调用方式：支持多种参数传递方式
- ✅ 完整类型提示：TypeScript 级别的类型安全
- ✅ 参考 APIFlask 标准：与框架保持一致

---

## 📦 安装位置

```
src/applications/common/utils/http.py
```

---

## 🎯 核心函数

### 1. `success()` - 成功响应

**签名：**
```python
def success(data: Any = None, message: str = '成功', code: int = 200) -> dict
```

**参数：**
- `data`：返回数据（任意类型）
- `message`：提示信息（默认 `'成功'`）
- `code`：业务状态码（默认 `200`）

**返回格式：**
```json
{
  "data": <返回数据>,
  "code": 200,
  "message": "成功"
}
```

**示例：**

```python
from iti.applications.common.utils import success
from iti.applications.extensions.http import BaseResponse

@app.get('/api/users/<int:user_id>')
@app.output(BaseResponse)
def get_user(user_id):
    user = User.query.get(user_id)
    if not user:
        return fail('用户不存在', code=404)
    
    # 返回单个对象
    return success({
        'id': user.id,
        'name': user.name,
        'email': user.email
    })

@app.get('/api/products')
@app.output(BaseResponse)
def get_products():
    products = Product.query.limit(10).all()
    
    # 返回列表
    return success(
        [{'id': p.id, 'name': p.name} for p in products],
        message='获取产品列表成功'
    )
```

---

### 2. `fail()` - 失败响应

**签名：**
```python
def fail(message: str = '操作失败', code: int = 500, data: Any = None) -> dict
```

**参数：**
- `message`：错误信息
- `code`：业务错误码（默认 `500`）
- `data`：额外数据（如验证错误详情）

**特点：**
- ⚠️ **HTTP 状态码保持 200**，由前端根据 `code` 字段判断业务状态
- 💡 适用于统一错误处理，避免 HTTP 层面的错误拦截

**返回格式：**
```json
{
  "data": null,
  "code": 500,
  "message": "操作失败"
}
```

**示例：**

```python
from iti.applications.common.utils import fail

@app.post('/api/users')
@app.output(BaseResponse)
def create_user():
    username = request.json.get('username')
    
    # 参数验证失败
    if not username:
        return fail('用户名不能为空', code=400)
    
    # 资源不存在
    if User.query.filter_by(username=username).first():
        return fail('用户名已存在', code=409)
    
    # 服务器错误
    try:
        user = User(username=username)
        db.session.add(user)
        db.session.commit()
    except Exception as e:
        return fail(f'创建失败: {str(e)}', code=500)
    
    return success(user_to_dict(user), message='创建成功')

@app.post('/api/login')
@app.output(BaseResponse)
def login():
    data = request.json
    
    # 验证失败，返回详细错误
    errors = validate_login(data)
    if errors:
        return fail(
            message='验证失败',
            code=422,
            data=errors  # {'username': ['必填项'], 'password': ['长度不足']}
        )
    
    # ... 登录逻辑
```

---

### 3. `page()` - 分页响应

**签名：**
```python
def page(
    items: Union[list, Any], 
    pagination: Union[dict, Any, None] = None, 
    message: str = '成功', 
    code: int = 200
) -> dict
```

**支持三种调用方式：**

#### **方式 1：传入 SQLAlchemy Pagination 对象（最简）**

```python
from iti.applications.common.utils import page

@app.get('/api/users')
@app.output(BaseResponse)
def get_users():
    page_num = request.args.get('page', 1, type=int)
    page_size = request.args.get('size', 10, type=int)
    
    # SQLAlchemy 分页查询
    db_pagination = User.query.paginate(page=page_num, per_page=page_size)
    
    # ✅ 直接传入 Pagination 对象，自动解析
    return page(db_pagination, message='获取用户列表成功')
```

**返回格式：**
```json
{
  "data": {
    "items": [
      {"id": 1, "name": "Alice"},
      {"id": 2, "name": "Bob"}
    ],
    "page": {
      "page": 1,
      "size": 10,
      "pages": 10,
      "total": 100,
      "current": "http://api.example.com/api/users?page=1&size=10",
      "next": "http://api.example.com/api/users?page=2&size=10",
      "prev": null,
      "first": "http://api.example.com/api/users?page=1&size=10",
      "last": "http://api.example.com/api/users?page=10&size=10"
    }
  },
  "code": 200,
  "message": "获取用户列表成功"
}
```

---

#### **方式 2：传入数据列表 + Pagination 对象**

适用于需要先序列化数据的场景：

```python
from iti.applications.common.utils import page
from iti.applications.schemas.user import UserSchema

@app.get('/api/users')
@app.output(BaseResponse)
def get_users():
    db_pagination = User.query.paginate(page=1, per_page=10)
    
    # 先序列化数据（使用 Marshmallow Schema）
    users = UserSchema(many=True).dump(db_pagination.items)
    
    # ✅ 传入序列化后的数据 + Pagination 对象
    return page(users, db_pagination, message='获取用户列表成功')
```

---

#### **方式 3：手动构建分页信息**

适用于非 SQLAlchemy 数据源（如 Redis、外部 API）：

```python
from iti.applications.common.utils import page, pagination_builder

@app.get('/api/stats')
@app.output(BaseResponse)
def get_stats():
    # 从 Redis 获取数据
    all_data = redis_client.lrange('stats', 0, -1)
    
    page_num = request.args.get('page', 1, type=int)
    page_size = request.args.get('size', 20, type=int)
    
    # 手动分页
    start = (page_num - 1) * page_size
    end = start + page_size
    items = all_data[start:end]
    
    # ✅ 手动构建分页信息
    pagination_info = pagination_builder(
        None,
        page=page_num,
        size=page_size,
        total=len(all_data)
    )
    
    return page(items, pagination_info, message='获取统计数据成功')
```

---

## 🔧 辅助函数

### `pagination_builder()` - 分页信息构建器

**签名：**
```python
def pagination_builder(
    pagination: Any,
    *,
    page: Optional[int] = None,
    size: Optional[int] = None,
    total: Optional[int] = None,
    pages: Optional[int] = None,
) -> dict
```

**参数说明：**
- `pagination`：SQLAlchemy Pagination 对象 或 `None`
- `page`：当前页码（手动模式）
- `size`：每页数量（手动模式）
- `total`：总记录数（手动模式）
- `pages`：总页数（可选，自动计算）

**注意：** 
- 第一个参数后使用 `*`，强制后续参数必须使用关键字传参
- 参考 APIFlask 的 `helpers.py` 实现

**示例 1：自动模式（SQLAlchemy Pagination）**

```python
from iti.applications.common.utils import pagination_builder

@app.get('/api/products')
def get_products():
    db_pagination = Product.query.paginate(page=1, per_page=10)
    
    # ✅ 自动从 Pagination 对象提取信息
    pagination_info = pagination_builder(db_pagination)
    
    return {
        'items': db_pagination.items,
        'pagination': pagination_info
    }
```

**示例 2：手动模式（自定义数据源）**

```python
from iti.applications.common.utils import pagination_builder

@app.get('/api/external-data')
def get_external_data():
    # 从外部 API 获取数据
    response = requests.get('https://api.example.com/data')
    total_count = response.headers.get('X-Total-Count', 0)
    items = response.json()
    
    # ✅ 手动构建分页信息
    pagination_info = pagination_builder(
        None,  # 第一个参数传 None
        page=1,
        size=20,
        total=int(total_count)
    )
    
    return page(items, pagination_info)
```

---

## 📊 Schema 定义

### `PaginationSchema` - 分页信息 Schema

```python
class PaginationSchema(Schema):
    """自定义分页信息 Schema"""
    page = Integer()      # 当前页码
    size = Integer()      # 每页数量（重命名自 per_page）
    pages = Integer()     # 总页数
    total = Integer()     # 总记录数
    current = URL()       # 当前页URL
    next = URL()          # 下一页URL
    prev = URL()          # 上一页URL
    first = URL()         # 首页URL
    last = URL()          # 末页URL
```

**特点：**
- ✅ 与 APIFlask 原生 `PaginationSchema` 保持一致
- ✅ 仅将 `per_page` 重命名为 `size`

---

### `PageDataSchema` - 分页数据 Schema

```python
class PageDataSchema(Schema):
    """分页数据包装 Schema"""
    items = List(Nested(Field()))  # 数据列表
    page = Nested(PaginationSchema)  # 分页信息（字段名为 page）
```

**使用示例：**

```python
from iti.applications.common.utils import PageDataSchema

# 在路由中使用（可选，主要用于文档生成）
@app.get('/api/users')
@app.doc(responses={200: PageDataSchema})
def get_users():
    ...
```

---

## 🎨 完整示例

### 示例 1：用户管理 CRUD

```python
from flask import request
from iti.applications.common.utils import success, fail, page
from iti.applications.extensions.http import BaseResponse
from my_project.models.user import User
from my_project.schemas.user import UserSchema, UserCreateSchema

# ========== 列表（分页） ==========
@app.get('/api/users')
@app.output(BaseResponse)
def get_users():
    """获取用户列表"""
    page_num = request.args.get('page', 1, type=int)
    page_size = request.args.get('size', 10, type=int)
    
    db_pagination = User.query.paginate(page=page_num, per_page=page_size)
    
    return page(db_pagination, message='获取用户列表成功')


# ========== 详情 ==========
@app.get('/api/users/<int:user_id>')
@app.output(BaseResponse)
def get_user(user_id):
    """获取用户详情"""
    user = User.query.get(user_id)
    if not user:
        return fail('用户不存在', code=404)
    
    return success(UserSchema().dump(user))


# ========== 创建 ==========
@app.post('/api/users')
@app.input(UserCreateSchema)
@app.output(BaseResponse)
def create_user(data):
    """创建用户"""
    # 检查用户名是否存在
    if User.query.filter_by(username=data['username']).first():
        return fail('用户名已存在', code=409)
    
    try:
        user = User(**data)
        db.session.add(user)
        db.session.commit()
        
        return success(
            UserSchema().dump(user),
            message='创建成功',
            code=201
        )
    except Exception as e:
        db.session.rollback()
        return fail(f'创建失败: {str(e)}', code=500)


# ========== 更新 ==========
@app.patch('/api/users/<int:user_id>')
@app.input(UserCreateSchema)
@app.output(BaseResponse)
def update_user(user_id, data):
    """更新用户"""
    user = User.query.get(user_id)
    if not user:
        return fail('用户不存在', code=404)
    
    try:
        for key, value in data.items():
            setattr(user, key, value)
        db.session.commit()
        
        return success(UserSchema().dump(user), message='更新成功')
    except Exception as e:
        db.session.rollback()
        return fail(f'更新失败: {str(e)}', code=500)


# ========== 删除 ==========
@app.delete('/api/users/<int:user_id>')
@app.output(BaseResponse)
def delete_user(user_id):
    """删除用户"""
    user = User.query.get(user_id)
    if not user:
        return fail('用户不存在', code=404)
    
    try:
        db.session.delete(user)
        db.session.commit()
        
        return success(None, message='删除成功')
    except Exception as e:
        db.session.rollback()
        return fail(f'删除失败: {str(e)}', code=500)
```

---

### 示例 2：复杂查询与筛选

```python
from sqlalchemy import and_, or_
from iti.applications.common.utils import page, pagination_builder

@app.get('/api/products')
@app.output(BaseResponse)
def get_products():
    """获取产品列表（支持筛选、搜索、排序）"""
    # 获取查询参数
    page_num = request.args.get('page', 1, type=int)
    page_size = request.args.get('size', 20, type=int)
    category = request.args.get('category')
    search = request.args.get('search')
    sort_by = request.args.get('sort', 'created_at')
    order = request.args.get('order', 'desc')
    
    # 构建查询
    query = Product.query
    
    # 筛选条件
    if category:
        query = query.filter(Product.category == category)
    
    # 搜索条件
    if search:
        query = query.filter(
            or_(
                Product.name.contains(search),
                Product.description.contains(search)
            )
        )
    
    # 排序
    if order == 'desc':
        query = query.order_by(getattr(Product, sort_by).desc())
    else:
        query = query.order_by(getattr(Product, sort_by).asc())
    
    # 分页
    db_pagination = query.paginate(page=page_num, per_page=page_size)
    
    # 序列化
    products = ProductSchema(many=True).dump(db_pagination.items)
    
    return page(products, db_pagination, message='获取产品列表成功')
```

---

### 示例 3：聚合统计（非 ORM 分页）

```python
from sqlalchemy import func
from iti.applications.common.utils import page, pagination_builder

@app.get('/api/stats/daily')
@app.output(BaseResponse)
def get_daily_stats():
    """获取每日统计数据"""
    page_num = request.args.get('page', 1, type=int)
    page_size = request.args.get('size', 30, type=int)
    
    # 聚合查询（不使用 ORM 分页）
    query = db.session.query(
        func.date(Order.created_at).label('date'),
        func.count(Order.id).label('order_count'),
        func.sum(Order.amount).label('total_amount')
    ).group_by(func.date(Order.created_at))
    
    # 获取总数
    total = query.count()
    
    # 手动分页
    offset = (page_num - 1) * page_size
    items = query.offset(offset).limit(page_size).all()
    
    # 格式化数据
    stats = [
        {
            'date': str(item.date),
            'order_count': item.order_count,
            'total_amount': float(item.total_amount or 0)
        }
        for item in items
    ]
    
    # 手动构建分页信息
    pagination_info = pagination_builder(
        None,
        page=page_num,
        size=page_size,
        total=total
    )
    
    return page(stats, pagination_info, message='获取统计数据成功')
```

---

## ⚙️ 配置说明

### BaseResponse Schema

在 `applications/extensions/http.py` 中定义：

```python
from apiflask import Schema
from apiflask.fields import Integer, String, Field

class BaseResponse(Schema):
    """统一响应格式 Schema"""
    data = Field()
    code = Integer()
    message = String()

def init_http(app):
    # 配置 APIFlask 使用自定义响应格式
    app.config["BASE_RESPONSE_SCHEMA"] = BaseResponse
    app.config["BASE_RESPONSE_DATA_KEY"] = "data"
```

---

## 🔍 URL 生成规则

分页 URL 自动生成逻辑：

1. **获取当前请求的 `base_url`**：`http://api.example.com/api/users`
2. **复制查询参数**：`request.args.copy()`
3. **更新分页参数**：
   - `page`: 页码
   - `size`: 每页数量（注意：使用 `size` 而非 `per_page`）
4. **构建完整 URL**：`base_url + ? + urlencode(args)`

**示例：**

请求 `/api/users?category=admin&page=2&size=10` 时生成的 URL：

```json
{
  "current": "http://api.example.com/api/users?category=admin&page=2&size=10",
  "next": "http://api.example.com/api/users?category=admin&page=3&size=10",
  "prev": "http://api.example.com/api/users?category=admin&page=1&size=10",
  "first": "http://api.example.com/api/users?category=admin&page=1&size=10",
  "last": "http://api.example.com/api/users?category=admin&page=10&size=10"
}
```

---

## 📝 最佳实践

### 1. 统一错误码规范

建议定义错误码常量：

```python
# applications/common/constants.py

class ErrorCode:
    """业务错误码"""
    SUCCESS = 200
    BAD_REQUEST = 400
    UNAUTHORIZED = 401
    FORBIDDEN = 403
    NOT_FOUND = 404
    CONFLICT = 409
    UNPROCESSABLE_ENTITY = 422
    INTERNAL_SERVER_ERROR = 500

# 使用
from iti.applications.common.utils import fail
from iti.applications.common.constants import ErrorCode

return fail('用户不存在', code=ErrorCode.NOT_FOUND)
```

---

### 2. 结合 Marshmallow Schema

```python
from iti.applications.schemas.user import UserSchema
from iti.applications.common.utils import success

@app.get('/api/users/<int:user_id>')
@app.output(BaseResponse)
def get_user(user_id):
    user = User.query.get_or_404(user_id)
    
    # ✅ 使用 Schema 序列化
    user_data = UserSchema().dump(user)
    
    return success(user_data)
```

---

### 3. 分页参数验证

```python
from iti.applications.common.utils import fail, page

@app.get('/api/users')
@app.output(BaseResponse)
def get_users():
    page_num = request.args.get('page', 1, type=int)
    page_size = request.args.get('size', 10, type=int)
    
    # 验证分页参数
    if page_num < 1:
        return fail('页码必须大于 0', code=400)
    
    if page_size < 1 or page_size > 100:
        return fail('每页数量必须在 1-100 之间', code=400)
    
    db_pagination = User.query.paginate(page=page_num, per_page=page_size)
    return page(db_pagination)
```

---

### 4. 异常统一处理

```python
from iti.applications.common.utils import fail

@app.errorhandler(404)
def handle_404(error):
    return fail('资源不存在', code=404)

@app.errorhandler(500)
def handle_500(error):
    return fail('服务器内部错误', code=500)

@app.errorhandler(Exception)
def handle_exception(error):
    app.logger.error(f'未处理的异常: {error}')
    return fail(str(error), code=500)
```

---

## 🆚 对比 APIFlask 原生实现

| 特性 | APIFlask 原生 | 本工具库 |
|------|--------------|---------|
| 分页参数名 | `per_page` | `size` ✅ |
| 返回格式 | 灵活 | 统一 `{data, code, message}` ✅ |
| 错误处理 | HTTP 状态码 | 业务 `code` 字段 ✅ |
| 智能识别 | 需手动处理 | 自动识别 Pagination 对象 ✅ |
| URL 生成 | 手动 | 自动生成 ✅ |

---

## 🧪 测试

### 运行测试

```bash
# 运行 HTTP 工具测试
uv run --extra dev pytest tests/test_http_utils.py -v

# 运行所有测试
uv run --extra dev pytest

# 测试覆盖率
uv run --extra dev pytest --cov=iti --cov-report=html
```

### 测试统计

- **测试用例数量**: 33 个
- **测试分类**:
  - `success()` 函数测试: 6 个
  - `fail()` 函数测试: 5 个
  - `pagination_builder()` 测试: 6 个
  - `page()` 函数测试: 7 个
  - Schema 定义测试: 2 个
  - Flask 集成测试: 4 个
  - 边界情况测试: 5 个

### 测试覆盖范围

- ✅ 基础功能测试
- ✅ 参数验证测试
- ✅ 默认值测试
- ✅ 边界情况测试
- ✅ Flask 应用集成测试
- ✅ Schema 定义测试
- ✅ 请求上下文处理测试

### 添加自定义测试

在 `tests/test_http_utils.py` 中添加测试：

```python
import pytest
from iti.applications.common.utils import success

def test_my_custom_case():
    """测试自定义场景"""
    result = success({'key': 'value'})
    assert result['code'] == 200
```

---

## 🐛 常见问题

### Q1: 为什么错误也返回 HTTP 200？

**A:** 这是一种常见的 API 设计模式，优点：
- 前端统一处理，不需要捕获 HTTP 异常
- 避免浏览器/代理对非 200 状态码的拦截
- 业务状态由 `code` 字段表示，更清晰

如需返回 HTTP 错误状态码，可以手动返回元组：
```python
return fail('未找到', code=404), 404  # HTTP 404
```

---

### Q2: 如何自定义分页 URL 生成逻辑？

**A:** 修改 `_generate_page_url()` 函数：

```python
def _generate_page_url(page_num, page_size):
    if page_num is None:
        return None
    
    # 自定义逻辑
    return f"https://custom-domain.com/api?p={page_num}&s={page_size}"
```

---

### Q3: 分页信息中的 URL 字段可以去掉吗？

**A:** 可以。修改 `pagination_builder()` 返回值：

```python
return {
    'page': page,
    'size': size,
    'pages': pages,
    'total': total,
    # 注释掉 URL 字段
    # 'current': current_url,
    # 'next': next_url,
    # ...
}
```

---

## 📚 参考资料

- [APIFlask 官方文档](https://apiflask.com/)
- [APIFlask helpers.py 源码](https://github.com/apiflask/apiflask/blob/main/src/apiflask/helpers.py)
- [Flask-SQLAlchemy Pagination](https://flask-sqlalchemy.palletsprojects.com/en/3.0.x/pagination/)

---

## 📅 更新日志

- **v1.0.0** (2024-10-14)
  - 初始版本
  - 实现 `success()`, `fail()`, `page()` 函数
  - 实现 `pagination_builder()` 分页构建器
  - 支持智能识别 SQLAlchemy Pagination 对象
  - 自动生成分页 URL
  - 参数重命名：`per_page` → `size`

---

**编写人员：** AI Assistant  
**最后更新：** 2024-10-14  
**版本：** 1.0.0