一、中间件 Middleware

定义

在每个 HTTP 请求进入路由处理函数（即你的 @app.get 函数）
本质是一个“环绕式（around）”的处理层。

作用

统一处理所有请求/响应的通用逻辑，例如：
- 记录请求日志（访问时间、IP、路径）
- 身份认证 / 权限校验
- CORS 跨域处理（FastAPI 有专门的 CORSMiddleware）
- 添加自定义响应头（如 X-Request-ID）
- 性能监控（记录处理耗时）

基本用法

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21


from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def middleware1(request: Request, call_next):
    print("中间件1 请求进入")
    response = await call_next(request)  # 调用下一个中间件或路由
    print("中间件1 响应返回")
    return response

@app.middleware("http")
async def middleware2(request: Request, call_next):
    print("中间件2 请求进入")
    response = await call_next(request)
    print("中间件2 响应返回")
    return response

@app.get("/")
async def root():
    return {"message": "Hello World!"}

执行顺序

中间件按代码定义顺序从上到下进入
但返回时从下到上退出

观察到的输出：

1
2
3
4
5


中间件1 请求进入
中间件2 请求进入
中间件2 响应返回
中间件1 响应返回
INFO: 127.0.0.1:54892 - "GET / HTTP/1.1" 200 OK

可以想象成“洋葱模型”：请求从外一层层进入，响应从内一层层返回。

参数具体详解

1
2
3
4
5
6


@app.middleware("http")
async def my_middleware(reques: Request, call_next):
    # 请求进入时的处理（before）
    response = await call_next(request)
    # 响应返回时的处理（after）
    return response

关键就是两个参数：

参数	类型	作用
`request`	`fastapi.Request`	当前 HTTP 请求对象，包含所有请求信息
`call_next`	可调用对象（`Callable`）	调用下一个处理单元（可能是另一个中间件，也可能是你的路由函数）

`request`：当前请求的完整信息包可以从中获取几乎所有请求细节：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23


from fastapi import Request

@app.middleware("http")
async def log_middleware(request: Request, call_next):
    # 获取请求方法
    method = request.method          # "GET", "POST"...
    
    # 获取 URL 路径
    url = request.url.path           # "/user/123"
    
    # 获取查询参数
    query_params = request.query_params  # {"limit": "10"}
    
    # 获取请求头
    headers = request.headers         # {"user-agent": "...", "authorization": "..."}
    
    # 获取客户端 IP（需注意代理）
    client_ip = request.client.host  # "127.0.0.1"
    
    print(f"[{method}] {url} from {client_ip}")
    
    response = await call_next(request)
    return response

可以把 request 看作 HTTP 请求的“只读快照”，用于读取和分析，但不能修改请求体 (也许可以修改但我不会)。

`call_next`：请求跳转的驱动核心

想象你点了一份外卖：

你（客户端）下单 →
平台（中间件1）记录订单 →
餐厅（中间件2）确认接单 →
厨师（你的路由函数）做菜 →
菜做好后，原路返回：厨师 → 餐厅 → 平台 → 你

call_next(request) 就是 把订单往下传 的动作。

技术解释：

call_next 是一个异步函数，它会：
1. 调用下一个中间件（如果还有）
2. 如果没有更多中间件，则调用最终的路由处理函数（如 @app.get("/")）
3. 等待它执行完毕，拿到 response 对象
4. 把 response 返回给你

所以：

await call_next(request) 之前的代码：请求进入阶段
await call_next(request) 之后的代码：响应返回阶段

完整示例：记录请求耗时

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15


import time
from fastapi import Request

@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    start_time = time.time()
    
    # 请求往下传（可能到下一个中间件，或路由函数）
    response = await call_next(request)
    
    # 响应回来了
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = f"{process_time:.4f}s"
    
    return response

这个中间件会在每个响应头中添加处理时间，比如：
X-Process-Time: 0.0123s

示例：拦截请求（不调用 `call_next`）

如果不调用 call_next，请求不会到达路由函数，也就是路由还没执行具体内容，直接停止，可以直接返回一个响应（比如拦截非法请求）

1
2
3
4
5
6
7
8


@app.middleware("http")
async def block_middleware(request: Request, call_next):
    if request.url.path == "/secret" and request.headers.get("x-key") != "123":
        # 直接返回 403，不调用 call_next
        return JSONResponse(status_code=403, content={"detail": "Forbidden"})
    
    # 否则继续
    return await call_next(request)

执行流程

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19


客户端
   ↓
[中间件1] 
   → 读取 request
   → 执行 before 逻辑
   → call_next() → 
        [中间件2]
           → 读取 request
           → 执行 before 逻辑
           → call_next() → 
                [路由函数] → 返回 response
           ← 拿到 response
           ← 执行 after 逻辑
           ← return response
   ← 拿到 response
   ← 执行 after 逻辑
   ← return response
   ↓
客户端

官方中间件示例：CORS

FastAPI 提供了常用中间件，无需手写：

1
2
3
4
5
6
7
8
9


from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 允许的前端域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

注意：app.add_middleware() 的执行顺序是先添加的先执行（与 @app.middleware 一致）。

二、依赖注入 Dependency Injection, DI

定义

一种解耦通用逻辑的设计模式。
FastAPI 的 DI 系统允许你将可复用的逻辑（如参数校验、数据库连接、用户认证）注入到路由函数中，而无需重复编写。

作用

避免代码重复：多个接口需要相同的查询参数（如分页 skip/limit）
逻辑集中管理：认证、数据库 session 等统一处理

基本用法

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17


from fastapi import Depends, Query

# 定义依赖函数
async def common_parameters(
    skip: int = Query(default=0, ge=0, le=100),
    limit: int = Query(default=10, ge=1, le=50),
):
    return {"skip": skip, "limit": limit}

# 在多个路由中注入
@app.get("/news/")
async def get_news_list(params: dict = Depends(common_parameters)):
    return {"news": [], **params}

@app.get("/users/")
async def get_users_list(params: dict = Depends(common_parameters)):
    return {"users": [], **params}

访问 /news/?skip=5&limit=20 → 自动校验 skip 和 limit 范围，并注入到函数中。

依赖注入的关键特点

特性	说明
类型安全	依赖函数的返回值类型可被类型注解捕获
嵌套依赖	依赖函数本身也可以使用 `Depends`
自动文档	依赖中的 `Query`, `Path`, `Body` 等参数会出现在 Swagger 中
灵活性	可用于获取当前用户、数据库连接、配置等

进阶示例：带认证的依赖

1
2
3
4
5
6
7
8


async def get_current_user(token: str = Header(...)):
    if token != "secret":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"user_id": 123}

@app.get("/profile")
async def profile(user: dict = Depends(get_current_user)):
    return user

所有需要登录的接口，只需加 Depends(get_current_user) 即可。

对比总结：中间件 vs 依赖注入

特性	中间件（Middleware）	依赖注入（Depends）
作用范围	全局（所有请求）	按需（指定路由）
执行时机	请求/响应的最外层	路由函数执行前
适用场景	日志、CORS、全局头	参数校验、用户认证、DB session
能否访问路径参数	不能直接获取（如 `/user/{id}` 中的 `id`）	可以（通过路由函数传入）
文档集成	不显示在 Swagger	自动显示参数