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

# 请求完整链路

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

## 请求链路总览

:::tip 链路概览
一个典型的 API 请求经过以下阶段：浏览器 → Nginx → CORS → 限流 → 上传体积限制 → 请求计时 → Token 续签 → 操作日志 → DB 会话 → Redis 注入 → 路由匹配 → JWT 认证 → RBAC 鉴权 → Pydantic 校验 → Endpoint → 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:8031;
    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 将请求代理到 FastAPI 服务，同时附加客户端真实 IP 到 `X-Forwarded-For` 头。

## 第二阶段：中间件链（洋葱模型外层 → 内层）

FastAPI 的中间件按注册顺序形成洋葱模型。请求从最外层进入，逐层穿透到路由处理函数，响应沿反向链返回。

### 3. CORS 跨域中间件（最外层）

```python
# src/middleware/cors.py — 最后注册，位于洋葱最外层
register_cors(app)
```

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

### 4. 请求限流中间件

```python
# src/middleware/rate_limit.py — 倒数第二注册，仅 CORS 内层
async def rate_limit_middleware(request: Request, call_next):
    if not RATE_LIMIT_ENABLED:
        return await call_next(request)
    # 滑动窗口计数器（Redis Lua 原子操作）
    count = await sliding_window_incr(redis, current_key, prev_key, ...)
    if count > RATE_LIMIT_LIMIT:
        return JSONResponse(
            status_code=200,
            content={
                "code": 1,
                "msg": "请求过于频繁，请稍后再试"
            }
        )
    return await call_next(request)
```

- 基于客户端 IP 的滑动窗口计数器
- 使用 Redis Lua 脚本保证原子性
- 超限直接返回 HTTP 200 + `{"code": 1}` 统一 JSON 响应，不进入后续链路
- 被限流的请求不进入 DB 会话与操作日志，避免 DoS 攻击时日志表与 DB 连接不降反增
- Redis 不可用时降级放行（fail-open）
- CORS 预检请求不计数

### 5. 上传体积限制中间件

```python
# src/middleware/upload_size_limit.py — 按 Content-Length 提前拒绝超大请求体
app.middleware("http")(upload_size_limit_middleware)
```

- 按 `Content-Length` 提前拒绝超大请求体
- 避免超大 multipart 落临时盘后再由业务层校验
- 位于限流内层，已被限流放行的请求才检查体积

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

```python
# src/core/app.py — 先注册→内层，统计业务耗时
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(round(process_time, 5))
    return response
```

- 记录请求处理耗时（不含限流/CORS 等外层开销）
- 通过 `X-Process-Time` 响应头返回

### 7. Token 续签中间件

```python
# src/core/app.py — 续签 token 写入响应头
async def token_refresh_middleware(request, call_next):
    response = await call_next(request)
    refresh_token = getattr(request.state, '_refresh_token', None)
    if refresh_token:
        response.headers['X-Refresh-Token'] = refresh_token
    return response
```

- 检查是否需要续签 Token
- 续签后的 Token 通过 `X-Refresh-Token` 响应头返回

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

```python
# src/middleware/operation_log.py — 在 DB 会话外层，记录写操作日志
async def operation_log_middleware(request: Request, call_next):
    response = await call_next(request)
    if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
        # 记录操作日志到 fastapi_operation_log 表
        # 包括：操作人、操作IP、请求路径、请求参数、响应状态、耗时等
        ...
    return response
```

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

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

```python
# src/middleware/db_session.py — 后注册→外层，DB 会话包裹业务逻辑
async def db_session_middleware(request: Request, call_next):
    # 创建数据库会话
    db = SessionLocal()
    # 注入到 contextvars，供业务层通过依赖注入获取
    _db_ctx.set(db)
    try:
        # 执行后续请求处理（路由、业务逻辑等）
        response = await call_next(request)
        # 对于写操作，根据业务返回码决定提交或回滚
        if request.method in ('POST', 'PUT', 'DELETE', 'PATCH'):
            body = await read_response_body(response)
            if _is_business_success(body):   # code == 0 表示业务成功
                db.commit()
            else:
                db.rollback()
        return response
    except Exception:
        # 发生异常时回滚事务
        db.rollback()
        raise
    finally:
        # 无论成功或异常，最终关闭会话释放连接
        db.close()
```

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

### 10. Redis 上下文注入中间件（最内层）

```python
# src/core/redis_client.py — 先注册→最内层：先注入 Redis 再开 DB 会话
async def redis_middleware(request: Request, call_next):
    # 从应用状态中获取 Redis 客户端实例
    redis = getattr(request.app.state, 'redis', None)
    if redis is not None:
        # 将 Redis 客户端注入到 contextvars，供业务层通过依赖注入获取
        _redis_ctx.set(redis)
    return await call_next(request)
```

- 将 Redis 异步客户端注入 `contextvars`
- 后续业务代码通过 `get_redis()` 获取当前请求的 Redis 客户端
- 位于最内层，确保 DB 会话和操作日志都能使用 Redis

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

### 11. 路由匹配

```python
# src/api/__init__.py
def register_router(app: FastAPI):
    # 登录和验证码无需认证，直接挂载到 app（同样带 /api/v1 前缀）
    app.include_router(login_view, prefix="/api/v1", tags=["系统登录"])

    # 需要认证的路由集合
    app.include_router(
        v1,
        prefix="/api/v1",
        dependencies=[Depends(login_required)],
    )
```

- FastAPI 根据 HTTP 方法 + URL 路径匹配到 `POST /api/v1/position/add` 端点
- 该端点挂载在 `v1` 路由器上，自动应用 `login_required` 依赖
- `v1` 路由器在 `src/api/v1/router.py` 中集中注册所有模块路由

### 12. JWT 认证（login_required）

```python
# src/api/deps.py
async def login_required(request: Request):
    # 1. 从 Authorization 头提取 Bearer Token
    # 2. 验证 Token 签名、有效期
    # 3. 检查 Token 是否在 Redis 黑名单中
    # 4. 解析出 user_id、username、realname
    # 5. 注入到 request.state
    request.state.user_id = user_id
    request.state.username = username
    request.state.realname = realname
```

- 认证失败抛出 `AuthorizationException`，由全局异常处理器返回统一响应
- 认证通过后，用户信息注入 `request.state` 供后续使用

### 13. RBAC 鉴权（permission_required）

```python
# src/api/v1/endpoints/position.py
@router.post('/add', summary='添加岗位')
@permission_required("sys:position:add")   # 权限校验：验证当前用户是否拥有该权限节点
@check_demo                                 # 演示模式校验：演示环境下禁止写操作
def add(request: Request, data: PositionForm):
    return position_service.add(request, data)
```

```python
# src/core/access_decorators.py
def permission_required(permission: str):
    """节点权限鉴权装饰器，userId==1（管理员）直接放行"""
    def decorator(func):
        def _check(request):
            """同步权限校验（直查DB，无Redis依赖，供线程池/回退场景使用）"""
            userId = get_user_id(request)
            if userId == 1:
                return None
            from modules.system.menu import service as menu
            permission_list = menu.get_permissions_list_sync(userId)
            if permission not in permission_list:
                return R.failed("权限不足")
            return None

        async def _check_cached(request):
            """异步权限校验（走 Redis 缓存，未命中才回源DB）"""
            userId = get_user_id(request)
            if userId == 1:
                return None
            from modules.system.menu import service as menu
            permission_list = await menu.get_permissions_list(userId)
            if permission not in permission_list:
                return R.failed("权限不足")
            return None

        if asyncio.iscoroutinefunction(func):
            @wraps(func)
            async def async_wrapper(*args, **kwargs):
                request = _extract_request(*args, **kwargs)
                denied = await _check_cached(request)
                if denied:
                    return denied
                return await func(*args, **kwargs)
            return async_wrapper

        @wraps(func)
        def sync_wrapper(*args, **kwargs):
            request = _extract_request(*args, **kwargs)
            denied = _check(request)
            if denied:
                return denied
            return func(*args, **kwargs)
        return sync_wrapper
    return decorator
```

- 权限字符串格式：`sys:{module}:{action}`（如 `sys:position:add`）
- 用户 ID=1（超级管理员）跳过所有权限检查
- 同步端点走 `_check` 直查 DB（在线程池中执行，不阻塞事件循环）
- 异步端点走 `_check_cached` 走 Redis 缓存查询，变更时主动失效
- 使用懒加载 `from modules.system.menu import service as menu` 避免 core → 业务模块顶层循环依赖

### 14. 演示模式校验（check_demo）

```python
# src/core/access_decorators.py
def check_demo(func):
    """演示模式拦截装饰器，放在 @permission_required 之后、路由函数之前"""
    if asyncio.iscoroutinefunction(func):
        @wraps(func)
        async def async_wrapper(*args, **kwargs):
            if FASTAPI_DEMO:
                return R.failed("演示环境，暂无操作权限")
            return await func(*args, **kwargs)
        return async_wrapper

    @wraps(func)
    def sync_wrapper(*args, **kwargs):
        if FASTAPI_DEMO:
            return R.failed("演示环境，暂无操作权限")
        return func(*args, **kwargs)
    return sync_wrapper
```

- `FASTAPI_DEMO=True` 时，所有写操作返回失败响应
- 根据被装饰函数类型自动选择同步/异步包装
- 保护演示环境数据不被修改

## 第四阶段：参数校验

### 15. Pydantic 请求校验

```python
# src/modules/system/position/schemas.py
class PositionForm(BaseForm):
    name: str = Field(..., min_length=1, max_length=150, description="岗位名称")
    status: int = Field(..., ge=1, le=2, description="岗位状态：1-在用 2-停用")
    sort: int = Field(..., ge=0, le=99999, description="岗位排序")
```

- FastAPI 自动解析 JSON 请求体
- Pydantic 按 `Field` 约束校验每个字段
- 校验失败触发 `RequestValidationError`，由全局异常处理器返回：

```json
{"code": 1, "data": null, "msg": "name: 确保该值至少有 1 个字符"}
```

- 校验通过后，`data` 参数为已验证的 `PositionForm` 实例

## 第五阶段：业务处理

### 16. Endpoint（HTTP 层）

```python
# src/api/v1/endpoints/position.py
def add(request: Request, data: PositionForm):
    return position_service.add(request, data)
```

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

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

```python
# src/modules/system/position/service.py
class PositionService(BaseService[Position]):
    repo = position_repo
    model = Position
    unique_fields = {'name': '岗位名称不能重复'}

    def add(self, request, data) -> R:
        # 1. 唯一性校验
        err = self._check_unique(data)
        if err:
            return R.failed(err)
        # 2. 前置扩展钩子
        self._before_add(request, data)
        # 3. 组装创建字段
        fields = self._build_create_fields(request, data)
        # 4. 调用 Repository 创建记录
        self.repo.create(**fields)
        # 5. 返回成功响应
        return R.ok(msg='添加成功')
```

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

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

### 19. Model（模型层）

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

### 20. 数据库执行

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

## 第六阶段：响应返回

### 21. 响应序列化

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

### 22. 中间件反向链

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

1. **请求计时**：附加 `X-Process-Time` 响应头
2. **Token 续签**：检查并附加 `X-Refresh-Token` 响应头
3. **操作日志**：记录写操作日志到数据库
4. **DB 会话**：检测 `code==0`，执行 `db.commit()`
5. **Redis 上下文**：无操作（contextvars 自动清理）
6. **限流**：无操作（响应直接透传）
7. **CORS**：附加 `Access-Control-Allow-*` 响应头

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

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

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

## 完整链路流程图

```
请求链路（自上而下，洋葱模型外层 → 内层）

[浏览器]
   │ POST /api/v1/position/add
   ▼
[Nginx] ── 反向代理转发
   │
   ▼
[CORS 中间件] ── 跨域校验通过（最外层）
   │
   ▼
[限流中间件] ── 滑动窗口计数（Redis Lua）
   │  ├─ 超限 → 直接返回 HTTP 200 + {code:1, msg:"请求过于频繁"}
   │  └─ 通过 ▼
   ▼
[上传体积限制] ── 按 Content-Length 拦截超大请求
   │
   ▼
[请求计时] ── 记录起始时间，响应时附加 X-Process-Time
   │
   ▼
[Token 续签] ── 检查是否需要续签，响应时附加 X-Refresh-Token
   │
   ▼
[操作日志中间件] ── 异步记录写操作（不影响主流程）
   │
   ▼
[DB 会话中间件] ── 创建 SessionLocal()，注入 contextvars
   │
   ▼
[Redis 注入中间件] ── 将 Redis 客户端注入 contextvars（最内层）
   │
   ▼
[路由层] ── 匹配 /api/v1/position/add
   │
   ▼
[认证依赖 login_required] ── JWT 验证 + 黑名单检查（Redis）
   │  ├─ 失败 → 返回 HTTP 200 + {code:401, msg:"..."}
   │  └─ 通过 → 解析 user_id/username 注入 request.state ▼
   ▼
[Endpoint 层]
   ├─ @permission_required ── 权限校验（同步直查DB / 异步Redis缓存）
   ├─ @check_demo ── 演示模式拦截
   └─ 调用 Service ▼
   ▼
[Service 层]
   ├─ _check_unique ── 唯一性校验（DB 查询）
   ├─ _before_add ── 前置扩展钩子
   ├─ _build_create_fields ── 组装创建字段
   └─ 调用 Repository ▼
   ▼
[Repository 层]
   ├─ _clean_model_data ── camelCase → snake_case，过滤非法字段
   └─ model.save() ── add + flush（获取自增 ID，不提交）▼
   ▼
[数据库] ── 执行 INSERT，返回自增 ID
   │
   ▼ 逐层返回响应（内层 → 外层）
[Redis] → [DBSession] → [OpLog] → [Token续签] → [请求计时] → [上传体积] → [限流] → [CORS]
   │
   ├─ 业务成功（code=0）→ DB 会话 commit 提交事务
   └─ 业务失败（code≠0）→ DB 会话 rollback 回滚事务
   │
   ▼
[Nginx]
   │
   ▼
[浏览器] ← JSON 响应（{ code, data, msg, ok }）
```

## 异常处理链路

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

| 异常类型 | 触发场景 | 响应（均为 HTTP 200） |
|:---|:---|:---|
| `AuthorizationException` | JWT 过期 / 无权限 | `{"code": 401, "msg": "..."}` |
| `BusinessException` | 业务逻辑错误 | `{"code": 1, "msg": "..."}` |
| `RequestValidationError` | Pydantic 校验失败 | `{"code": 1, "msg": "field: message"}` |
| `StarletteHTTPException` | 404 / 405 等 | `{"code": 1, "msg": "请求错误"}` |
| `Exception` | 未捕获异常 | `{"code": 1, "msg": "服务器内部错误"}` |

:::warning 异常处理原则
所有异常均返回 **HTTP 200** 状态码，通过响应体 `code` 字段区分业务成功（0）与失败（1/401 等）。注意：`AuthorizationException` 的 `code=401` 是响应体中的业务码，不是 HTTP 状态码。这是前后端约定，前端统一按 `code` 判断，无需处理 HTTP 状态码差异。
:::

## 总结

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