企业官网建设流程全解析

依赖注入系统（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 会：

1. 先调用get_current_user()
2. 把返回值传给current_user参数
3. 再执行你的路由函数

三、常见场景 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 会：

1. 调用get_api_key
2. 用其结果调用get_current_user_from_key
3. 最后执行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) 中：

1. 点击任意受保护接口
2. 点击 “Authorize”
3. 输入valid-token-123
4. 即可正常测试

八、练习任务（动手实践）

🧠 请先自己尝试完成，再查看下方答案！

任务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 的接口