djangoadmin-wiki/DjangoAdmin_Flask_EleVue/zh/volume3/request-flow.md

# 请求完整链路

本章以一个具体的 API 请求为例，详细描述从浏览器发起 HTTP 请求到返回响应的完整链路，帮助开发者理解每一层的职责和数据流转方式。

## 请求链路总览

:::tip 链路概览
一个典型的 API 请求经过以下阶段：浏览器 → Nginx → CORS → Redis 注入 → DB 会话 → 操作日志 → 路由匹配 → JWT 认证 → RBAC 鉴权 → WTForms 校验 → Router → Service → Repository → DB → 响应序列化 → 中间件反向链 → 浏览器
:::

以下以 `POST /api/v1/position/add` 添加岗位为例，逐步拆解完整链路。

## 第一阶段：网络层

### 1. 浏览器发起请求

```javascript
// 前端 Axios 调用
const res = await request({
    url: '/api/v1/position/add',
    method: 'post',
    data: {
        name: '高级工程师',
        status: 1,
        sort: 10
    }
});
```

前端通过 Axios 发起 HTTP POST 请求，携带：
- `Authorization: Bearer <JWT Token>` 请求头
- `Content-Type: application/json` 请求头
- JSON 请求体

### 2. Nginx 反向代理（生产环境）

```nginx
location /api/ {
    proxy_pass http://127.0.0.1:8021;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}
```

Nginx 将请求代理到 Flask 服务，同时附加客户端真实 IP 到 `X-Forwarded-For` 头。

## 第二阶段：中间件链（请求/响应钩子）

Flask 的中间件通过 `before_request` / `after_request` 钩子实现。请求从最外层进入，逐层穿透到路由处理函数，响应沿反向链返回。

### 3. CORS 跨域配置（最外层）

```python
# src/app.py - create_app()
from flask_cors import CORS
from config import CORS_ALLOWED_ORIGINS

if CORS_ALLOWED_ORIGINS:
    allowed_origins = [origin.strip() for origin in CORS_ALLOWED_ORIGINS.split(',') if origin.strip()]
    CORS(app, resources={r"/*": {"origins": allowed_origins}}, supports_credentials=True)
else:
    CORS(app, resources={r"/*": {"origins": "*"}}, supports_credentials=False)
```

- 检查 `Origin` 头是否在允许列表中
- 预检请求（OPTIONS）直接返回 CORS 响应头
- 普通请求附加 `Access-Control-Allow-*` 响应头

### 4. Redis 上下文注入中间件

```python
# src/core/redis_client.py
@app.before_request
def redis_context():
    # 将 Redis 客户端注入到 g 对象
    g.redis = redis_client
```

- 将 Redis 客户端注入 `g` 对象
- 后续业务代码通过 `get_redis()` 获取当前请求的 Redis 客户端

### 6. 数据库会话中间件

```python
# src/middleware/transaction.py
@app.before_request
def _transaction_begin():
    """请求开始前：标记事务活跃"""
    g.transaction_active = True

@app.after_request
def _transaction_commit(response):
    """请求结束后：根据响应状态决定提交或回滚"""
    if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
        if response.status_code == 200:
            body = response.get_json()
            if body and body.get('code') == 0:
                db.session.commit()
            else:
                db.session.rollback()
    return response

@app.teardown_request
def _transaction_rollback(exception):
    """请求异常时：回滚事务"""
    if exception:
        db.session.rollback()
    db.session.remove()
```

- 为每个请求管理数据库会话
- 写操作根据响应体 `code` 字段自动 commit/rollback
- 异常时自动 rollback，finally 中关闭会话

### 7. 操作日志中间件

```python
# src/middleware/operation_log.py
@app.after_request
def operation_log(response):
    if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
        # 记录操作日志到 flask_operation_log 表
        # 包括：操作人、操作IP、请求路径、请求参数、响应状态、耗时等
        ...
    return response
```

- 在 DB 会话外层，记录写操作日志
- 包含请求方法、URL、参数、响应码、耗时、操作人等信息

### 8. 请求计时中间件

```python
# src/app.py
@app.before_request
def start_timer():
    g.start_time = time.time()

@app.after_request
def add_process_time_header(response):
    if hasattr(g, 'start_time'):
        process_time = time.time() - g.start_time
        response.headers["X-Process-Time"] = str(round(process_time, 5))
    return response
```

- 记录请求处理耗时
- 通过 `X-Process-Time` 响应头返回

## 第三阶段：路由与认证

### 9. 路由匹配

```python
# src/routes/v1.py
def register_v1(app):
    v1 = Blueprint('v1', __name__, url_prefix='/api/v1')
    for blue in BLUEPRINTS:
        v1.register_blueprint(blue)
    app.register_blueprint(v1)
```

- Flask 根据 HTTP 方法 + URL 路径匹配到 `POST /api/v1/position/add` 端点
- 该端点挂载在 `v1` 蓝图上
- `v1` 蓝图在 `src/routes/v1.py` 中集中注册所有模块路由

### 10. JWT 认证（login_required）

```python
# src/middleware/auth.py
@app.before_request
def login_required():
    # 白名单路径跳过认证
    if request.path in LOGIN_WHITELIST:
        return

    # 1. 从 Authorization 头提取 Bearer Token
    # 2. 验证 Token 签名、有效期
    # 3. 检查 Token 是否在 Redis 黑名单中
    # 4. 解析出 user_id、username、realname
    # 5. 注入到 g 对象
    g.user_id = user_id
    g.username = username
    g.realname = realname
```

- 认证失败返回 `{"code": 401, "msg": "..."}` 响应
- 认证通过后，用户信息注入 `g` 对象供后续使用

### 11. RBAC 鉴权（check_permission）

```python
# src/modules/system/position/router.py
@position_blue.route('/add', methods=['POST'])
@check_permission('sys:position:add')
def add():
    from src.modules.system.position.service import add_position
    from src.modules.system.position.schema import PositionForm
    data = parse_request_body(request, PositionForm)
    return add_position(data)
```

```python
# src/core/permission.py
def check_permission(permission: str):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            user_id = get_user_id()
            if user_id == 1:
                return func(*args, **kwargs)
            from src.modules.system.menu.service import get_permissions_list
            permission_list = get_permissions_list(user_id)
            if permission not in permission_list:
                return R.failed("权限不足")
            return func(*args, **kwargs)
        return wrapper
    return decorator
```

- 权限字符串格式：`sys:{module}:{action}`（如 `sys:position:add`）
- 用户 ID=1（超级管理员）跳过所有权限检查
- 使用懒加载 `from src.modules.system.menu.service import get_permissions_list` 避免 core → 业务模块顶层循环依赖

## 第四阶段：参数校验

### 12. WTForms 请求校验

```python
# src/modules/system/position/schema.py
class PositionForm(BaseForm):
    name = StringField('岗位名称', validators=[
        DataRequired(message='岗位名称不能为空'),
        Length(min=1, max=150, message='岗位名称长度必须在1-150个字符之间')
    ])
    status = IntegerField('岗位状态', validators=[
        DataRequired(message='岗位状态不能为空'),
        NumberRange(min=1, max=2, message='岗位状态必须是1或2')
    ])
    sort = IntegerField('岗位排序', validators=[
        DataRequired(message='岗位排序不能为空'),
        NumberRange(min=0, max=99999, message='岗位排序必须在0-99999之间')
    ])
```

- `parse_request_body` 函数解析 JSON 请求体
- WTForms 按 `validators` 约束校验每个字段
- 校验失败返回：

```json
{"code": 1, "data": null, "msg": "岗位名称不能为空"}
```

- 校验通过后，`data` 参数为已验证的字典

## 第五阶段：业务处理

### 13. Router（HTTP 层）

```python
# src/modules/system/position/router.py
def add():
    from src.modules.system.position.service import add_position
    from src.modules.system.position.schema import PositionForm
    data = parse_request_body(request, PositionForm)
    return add_position(data)
```

- HTTP 层仅做请求转发，不含业务逻辑
- 直接调用 Service 层函数

### 14. Service（业务逻辑层）

```python
# src/modules/system/position/service.py
class PositionService(BaseService):
    repository = PositionRepository
    form = PositionForm
    status_form = PositionStatusForm
    model_name = "岗位"

    @classmethod
    def before_delete(cls, ids):
        """删除前校验"""
        for pid in ids:
            if UserRepository.count_by({'position_id': pid}) > 0:
                raise BizError(f"岗位ID {pid} 已被用户引用，无法删除")

# 模块级代理函数
add_position = PositionService.add
```

### 15. Repository（数据访问层）

```python
# src/common/base_repository.py
class BaseRepository:
    def create(self, **fields):
        # 清洗数据：camelCase → snake_case，过滤非模型字段
        obj = cls.model(**cls._clean_form_data(fields))
        obj.save()  # add + flush（不 commit）
        return obj
```

### 16. Model（模型层）

```python
# src/common/base_db.py
class BaseDB:
    def save(self):
        db.session.add(self)
        db.session.flush()  # 发送 SQL 到数据库，获取自增 ID
        return self
```

### 17. 数据库执行

- SQLAlchemy 将 ORM 操作转换为 SQL：`INSERT INTO flask_position (name, status, sort, create_user, ...) VALUES (...)`
- `flush()` 将 SQL 发送到数据库，获取自增主键
- 事务由中间件在响应返回后统一 commit

## 第六阶段：响应返回

### 18. 响应序列化

```python
# Service 返回 R.ok(msg='添加成功')
# src/core/response.py
R.ok(msg='添加成功')
# → jsonify({"code": 0, "data": None, "msg": "添加成功", "ok": True})
```

### 19. 中间件反向链

响应沿中间件链反向返回：

1. **请求计时**：附加 `X-Process-Time` 响应头
2. **操作日志**：记录写操作日志到数据库
3. **DB 会话**：检测 `code==0`，执行 `db.session.commit()`
4. **Redis 上下文**：无操作（g 对象自动清理）
5. **CORS**：附加 `Access-Control-Allow-*` 响应头

### 20. 浏览器接收响应

```json
{
    "code": 0,
    "data": null,
    "msg": "添加成功",
    "ok": true
}
```

前端 Axios 拦截器统一处理响应，`code===0` 时展示成功提示，否则展示错误信息。

## 完整链路流程图

```
请求链路（自上而下）

[浏览器]
   │ POST /api/v1/position/add
   ▼
[Nginx] ── 反向代理转发
   │
   ▼
[CORS 中间件] ── 跨域校验通过
   │
   ▼
[DB 会话中间件] ── 标记事务活跃
   │
   ▼
[操作日志中间件] ── 记录写操作（不影响主流程）
   │
   ▼
[路由层] ── 匹配 /api/v1/position/add
   │
   ▼
[认证中间件 login_required] ── JWT 验证 + 黑名单检查（Redis）
   │  ├─ 失败 → 返回 401
   │  └─ 通过 → 解析 user_id/username 注入 g 对象 ▼
   ▼
[Router 层]
   ├─ @check_permission ── 权限校验（Redis 缓存）
   ├─ parse_request_body ── WTForms 参数校验
   └─ 调用 Service ▼
   ▼
[Service 层]
   ├─ _validate_form ── WTForms 表单校验
   ├─ before_add ── 前置扩展钩子
   ├─ _clean_form_data ── 过滤无效字段
   └─ 调用 Repository.create ▼
   ▼
[Repository 层]
   └─ model(**kwargs).save() ── add + flush（获取自增 ID，不提交）▼
   ▼
[数据库] ── 执行 INSERT，返回自增 ID
   │
   ▼ 逐层返回响应
[Repository] → [Service] → [Router] → [OpLog] → [DBSession]
   │
   ├─ 业务成功（code=0）→ commit 提交事务
   └─ 业务失败（code≠0）→ rollback 回滚事务
   │
   ▼
[CORS] → [Nginx]
   │
   ▼
[浏览器] ← JSON 响应（{ code, data, msg, ok }）
```

## 异常处理链路

当请求链路中发生异常时，全局异常处理器统一捕获并返回标准响应：

| 异常类型 | 触发场景 | 响应 |
|:---|:---|:---|
| `BizError` | 业务逻辑错误 | `{"code": 1, "msg": "..."}` |
| `ValidationError` | WTForms 校验失败 | `{"code": 1, "msg": "field: message"}` |
| `Exception` | 未捕获异常 | `{"code": 1, "msg": "服务器内部错误"}` |

:::warning 异常处理原则
所有异常均返回 HTTP 200 状态码，通过 `code` 字段区分业务成功（0）与失败（1）。这是前后端约定，前端统一按 `code` 判断，无需处理 HTTP 状态码差异。
:::

## 总结

一个 API 请求从浏览器到数据库经过 7 个阶段、20+ 个处理节点。中间件链负责横切关注点（安全、日志、事务），路由层负责认证鉴权，WTForms 负责参数校验，三层架构负责业务逻辑与数据访问。这种分层设计使得每一层职责单一、可独立测试，系统在高并发场景下仍能保持稳定的性能和可靠的事务一致性。