依赖注入系统（Dependency Injection）
💡 本部分目标：理解 FastAPI 强大的依赖注入机制，学会复用逻辑（如认证、数据库连接、配置加载），让代码更模块化、可测试、可维护。
一、什么是“依赖注入”（Dependency Injection）？
在编程中，“依赖”是指一个函数或类需要的外部资源或服务，比如：
- 数据库连接
- 当前登录用户信息
- API 密钥验证
- 配置参数
依赖注入（DI）就是：在运行时自动提供这些依赖，而不是在函数内部硬编码创建它们。
FastAPI 的依赖注入有什么好处？
- 代码复用：多个接口共享同一逻辑（如验证 token）
- 解耦：业务逻辑与基础设施（如数据库）分离
- 可测试：可以轻松替换依赖（如用 mock 数据库）
- 声明式：只需在函数参数中声明
Depends(...)，FastAPI 自动处理
二、基础用法：使用Depends()
FastAPI 通过fastapi.Depends实现依赖注入。
步骤 1：定义一个“依赖函数”
defget_current_user():# 模拟从 token 中解析用户（实际项目会验证 JWT）return{"username":"alice","role":"admin"}步骤 2：在路径操作函数中使用它
fromfastapiimportDepends,FastAPI app=FastAPI()@app.get("/profile")defread_profile(current_user=Depends(get_current_user)):return{"user":current_user}🔍 当访问
/profile时，FastAPI 会：
- 先调用
get_current_user()- 把返回值传给
current_user参数- 再执行你的路由函数
三、常见场景 1：API Key 验证
很多 API 要求客户端在请求头中携带X-API-Key。
示例：验证 API Key
fromfastapiimportDepends,FastAPI,HTTPException,Header app=FastAPI()# 依赖函数：验证 API Keydefverify_api_key(x_api_key:str=Header(...)):valid_keys=["secret123","mykey456"]ifx_api_keynotinvalid_keys:raiseHTTPException(status_code=403,detail="Invalid API Key")returnx_api_key# 使用依赖@app.get("/protected-data",dependencies=[Depends(verify_api_key)])defget_protected_data():return{"data":"这是敏感数据！"}⚠️ 注意：
Header(...)表示从请求头中读取x-api-key（FastAPI 会自动转为小写和下划线）- 使用
dependencies=[Depends(...)]表示只执行依赖，不接收返回值
测试方式：
- 请求头中添加：
X-API-Key: secret123 - 如果 key 无效，返回 403 Forbidden
四、常见场景 2：数据库会话管理
在真实项目中，每个请求通常需要一个数据库会话，并在结束后关闭。
示例：模拟数据库连接
fromfastapiimportDepends,FastAPI# 模拟数据库classFakeDB:def__init__(self):self.connected=Truedefquery(self,sql:str):returnf"执行:{sql}"defclose(self):self.connected=False# 依赖函数：提供数据库会话defget_db():db=FakeDB()try:yielddb# 提供 db 给路由函数finally:db.close()# 请求结束后自动关闭print("数据库连接已关闭")app=FastAPI()@app.get("/items/")defread_items(db=Depends(get_db)):result=db.query("SELECT * FROM items")return{"result":result}🔑 关键点：
- 使用
yield实现“请求开始时创建，结束时清理”- 即使发生异常，
finally块也会执行，确保资源释放
五、子依赖（嵌套依赖）
依赖函数本身也可以依赖其他依赖！
示例：先验证 API Key，再获取用户
defget_api_key(x_api_key:str=Header(...)):ifx_api_key!="supersecret":raiseHTTPException(status_code=403,detail="Invalid key")returnx_api_keydefget_current_user_from_key(api_key:str=Depends(get_api_key)):# 根据 api_key 查找用户（简化）return{"user_id":"U123","api_key":api_key}@app.get("/user-info")defuser_info(user=Depends(get_current_user_from_key)):returnuser调用/user-info时，FastAPI 会：
- 调用
get_api_key - 用其结果调用
get_current_user_from_key - 最后执行
user_info
六、依赖注入 vs 普通函数调用？
| 方式 | 优点 | 缺点 |
|---|---|---|
| 普通函数调用 | 简单直接 | 无法复用、难以测试、无法自动处理异常/清理 |
| 依赖注入 | 自动管理生命周期、支持嵌套、可复用、可测试 | 初学稍复杂 |
最佳实践：凡是需要跨多个接口复用或需要资源管理的逻辑，都应写成依赖。
七、完整示例代码（推荐保存为main.py）
# main.pyfromfastapiimportDepends,FastAPI,HTTPException,HeaderfromtypingimportGenerator app=FastAPI(title="第4部分：依赖注入系统")# === 1. 模拟数据库 ===classDatabaseSession:def__init__(self):self.id="DB_SESSION_001"defexecute(self,query:str):returnf"[{self.id}] 执行:{query}"defclose(self):print(f"[{self.id}] 数据库连接已关闭")defget_db()->Generator[DatabaseSession,None,None]:db=DatabaseSession()try:yielddbfinally:db.close()# === 2. API Key 验证 ===defverify_token(x_token:str=Header(...)):ifx_token!="valid-token-123":raiseHTTPException(status_code=401,detail="无效的 Token")returnx_token# === 3. 获取当前用户（依赖 token）===defget_current_user(token:str=Depends(verify_token)):return{"username":"alice","token":token}# === 路由 ===@app.get("/health")defhealth_check():return{"status":"OK"}@app.get("/items/",dependencies=[Depends(verify_token)])deflist_items(db=Depends(get_db)):data=db.execute("SELECT name FROM items")return{"items":[data]}@app.get("/profile")defprofile(user=Depends(get_current_user)):return{"profile":user}测试说明：
- 访问
/health→ 无需 token - 访问
/items/或/profile→ 必须在请求头中添加：X-Token: valid-token-123
在 Swagger UI (/docs) 中：
- 点击任意受保护接口
- 点击 “Authorize”
- 输入
valid-token-123 - 即可正常测试
八、练习任务（动手实践）
🧠 请先自己尝试完成，再查看下方答案！
任务1：实现“日志记录”依赖
- 创建一个依赖函数
log_request - 每次被调用时，打印
"收到请求: {path}" - 在
/logs路由中使用它（不接收返回值）
任务2：组合依赖 —— 验证角色
- 依赖1：
get_current_user()返回{"username": "bob", "role": "user"} - 依赖2：
require_admin(user=Depends(get_current_user))
如果user["role"] != "admin"，抛出 403 错误 - 路由
/admin-only使用require_admin依赖
任务3（挑战）：带参数的依赖
- 创建依赖
get_pagination(skip: int = 0, limit: int = 10) - 返回字典
{"skip": skip, "limit": limit} - 在
/products/路由中使用它，并返回分页信息
九、练习任务参考答案
任务1 答案
fromfastapiimportRequestdeflog_request(request:Request):print(f"收到请求:{request.url.path}")@app.get("/logs",dependencies=[Depends(log_request)])deflogs_endpoint():return{"message":"请求已记录"}💡
Request对象由 FastAPI 自动注入，可获取 URL、方法、头等信息。
任务2 答案
defget_current_user():# 模拟当前用户（实际项目从 token 解析）return{"username":"bob","role":"user"}defrequire_admin(user=Depends(get_current_user)):ifuser["role"]!="admin":raiseHTTPException(status_code=403,detail="需要管理员权限")returnuser@app.get("/admin-only",dependencies=[Depends(require_admin)])defadmin_only():return{"message":"欢迎，管理员！"}🔒 访问
/admin-only会返回 403，因为模拟用户是"role": "user"。
可修改get_current_user返回"role": "admin"来测试成功情况。
任务3 答案
defget_pagination(skip:int=0,limit:int=10):return{"skip":skip,"limit":limit}@app.get("/products/")deflist_products(pagination=Depends(get_pagination)):return{"message":"产品列表","pagination":pagination}测试：
/products/→{"skip":0, "limit":10}/products/?skip=20&limit=5→{"skip":20, "limit":5}
十、小结
在本部分，你学会了：
- 使用
Depends()注入依赖函数 - 实现API 验证、数据库会话管理、角色权限控制
- 使用
yield实现资源自动清理 - 构建嵌套依赖（子依赖）
- 在 Swagger 中测试带 Header 的接口
