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

请求完整链路

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

请求链路总览

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

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

第一阶段：网络层

1. 浏览器发起请求

// 前端 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 反向代理（生产环境）

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 跨域中间件（最外层）

# src/middleware/cors.py
register_cors(app) # 注册 CORS 中间件

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

4. 上传体积限制中间件

# src/middleware/upload_size.py
app.middleware("http")(upload_size_limit_middleware)

• 按 Content-Length 提前拒绝超大请求体
• 避免超大 multipart 落临时盘后再由业务层校验

5. 请求限流中间件

# src/middleware/rate_limit.py
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 脚本保证原子性
• 超限直接返回统一 JSON 响应，不进入后续链路
• Redis 不可用时降级放行（fail-open）

6. Redis 上下文注入中间件

# src/core/redis_client.py
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 客户端

7. 数据库会话中间件

# src/middleware/db_session.py
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 中关闭会话

8. 操作日志中间件

# src/middleware/operation_log.py
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. Token 续签中间件

# src/core/app.py
async def token_refresh_middleware(request: 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 响应头返回

10. 请求计时中间件

# 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

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

第三阶段：路由与认证

11. 路由匹配

# 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）

# 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）

# 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)

# 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）

# src/core/access_decorators.py
def check_demo(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        if FASTAPI_DEMO:
            return R.failed("演示环境，暂无操作权限")
        return await func(*args, **kwargs)
    return wrapper

• FASTAPI_DEMO=True 时，所有写操作返回失败响应
• 保护演示环境数据不被修改

第四阶段：参数校验

15. Pydantic 请求校验

# 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，由全局异常处理器返回：

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

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

第五阶段：业务处理

16. Endpoint（HTTP 层）

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

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

17. Service（业务逻辑层）

# 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（数据访问层）

# 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（模型层）

# 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. 响应序列化

# 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. 浏览器接收响应

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

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

完整链路流程图

请求链路（自上而下）

[浏览器]
   │ POST /api/v1/position/add
   ▼
[Nginx] ── 反向代理转发
   │
   ▼
[CORS 中间件] ── 跨域校验通过
   │
   ▼
[限流中间件] ── 滑动窗口计数（Redis）
   │  ├─ 超限 → 直接返回 429
   │  └─ 通过 ▼
   ▼
[DB 会话中间件] ── 创建 SessionLocal()，注入 contextvars
   │
   ▼
[操作日志中间件] ── 异步记录写操作（不影响主流程）
   │
   ▼
[路由层] ── 匹配 /api/v1/position/add
   │
   ▼
[认证依赖 login_required] ── JWT 验证 + 黑名单检查（Redis）
   │  ├─ 失败 → 返回 401
   │  └─ 通过 → 解析 user_id/username 注入 request.state ▼
   ▼
[Endpoint 层]
   ├─ @check_demo ── 演示模式拦截
   ├─ @permission_required ── 权限校验（Redis 缓存）
   └─ 调用 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
   │
   ▼ 逐层返回响应
[Repository] → [Service] → [Endpoint] → [Router] → [OpLog] → [DBSession]
   │
   ├─ 业务成功（code=0）→ commit 提交事务
   └─ 业务失败（code≠0）→ rollback 回滚事务
   │
   ▼
[限流] → [CORS] → [Nginx]
   │
   ▼
[浏览器] ← JSON 响应（{ code, data, msg, ok }）

异常处理链路

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

异常类型	触发场景	响应
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）。这是前后端约定，前端统一按 code 判断，无需处理 HTTP 状态码差异。 :::

总结

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