2026/5/21 12:29:12
网站建设
项目流程
中国建设招标网 官方网站,网站制作的基本流程,wordpress打开速度慢解决办法,遵义做网站推广依赖注入系统#xff08;Dependency Injection#xff09;#x1f4a1; 本部分目标#xff1a;理解 FastAPI 强大的依赖注入机制#xff0c;学会复用逻辑#xff08;如认证、数据库连接、配置加载#xff09;#xff0c;让代码更模块化、可测试、可维护。一、什么是“依…依赖注入系统Dependency Injection 本部分目标理解 FastAPI 强大的依赖注入机制学会复用逻辑如认证、数据库连接、配置加载让代码更模块化、可测试、可维护。一、什么是“依赖注入”Dependency Injection在编程中“依赖”是指一个函数或类需要的外部资源或服务比如数据库连接当前登录用户信息API 密钥验证配置参数依赖注入DI就是在运行时自动提供这些依赖而不是在函数内部硬编码创建它们。FastAPI 的依赖注入有什么好处代码复用多个接口共享同一逻辑如验证 token解耦业务逻辑与基础设施如数据库分离可测试可以轻松替换依赖如用 mock 数据库声明式只需在函数参数中声明Depends(...)FastAPI 自动处理二、基础用法使用Depends()FastAPI 通过fastapi.Depends实现依赖注入。步骤 1定义一个“依赖函数”defget_current_user():# 模拟从 token 中解析用户实际项目会验证 JWTreturn{username:alice,role:admin}步骤 2在路径操作函数中使用它fromfastapiimportDepends,FastAPI appFastAPI()app.get(/profile)defread_profile(current_userDepends(get_current_user)):return{user:current_user} 当访问/profile时FastAPI 会先调用get_current_user()把返回值传给current_user参数再执行你的路由函数三、常见场景 1API Key 验证很多 API 要求客户端在请求头中携带X-API-Key。示例验证 API KeyfromfastapiimportDepends,FastAPI,HTTPException,Header appFastAPI()# 依赖函数验证 API Keydefverify_api_key(x_api_key:strHeader(...)):valid_keys[secret123,mykey456]ifx_api_keynotinvalid_keys:raiseHTTPException(status_code403,detailInvalid API Key)returnx_api_key# 使用依赖app.get(/protected-data,dependencies[Depends(verify_api_key)])defget_protected_data():return{data:这是敏感数据}⚠️ 注意Header(...)表示从请求头中读取x-api-keyFastAPI 会自动转为小写和下划线使用dependencies[Depends(...)]表示只执行依赖不接收返回值测试方式请求头中添加X-API-Key: secret123如果 key 无效返回 403 Forbidden四、常见场景 2数据库会话管理在真实项目中每个请求通常需要一个数据库会话并在结束后关闭。示例模拟数据库连接fromfastapiimportDepends,FastAPI# 模拟数据库classFakeDB:def__init__(self):self.connectedTruedefquery(self,sql:str):returnf执行:{sql}defclose(self):self.connectedFalse# 依赖函数提供数据库会话defget_db():dbFakeDB()try:yielddb# 提供 db 给路由函数finally:db.close()# 请求结束后自动关闭print(数据库连接已关闭)appFastAPI()app.get(/items/)defread_items(dbDepends(get_db)):resultdb.query(SELECT * FROM items)return{result:result} 关键点使用yield实现“请求开始时创建结束时清理”即使发生异常finally块也会执行确保资源释放五、子依赖嵌套依赖依赖函数本身也可以依赖其他依赖示例先验证 API Key再获取用户defget_api_key(x_api_key:strHeader(...)):ifx_api_key!supersecret:raiseHTTPException(status_code403,detailInvalid key)returnx_api_keydefget_current_user_from_key(api_key:strDepends(get_api_key)):# 根据 api_key 查找用户简化return{user_id:U123,api_key:api_key}app.get(/user-info)defuser_info(userDepends(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 appFastAPI(title第4部分依赖注入系统)# 1. 模拟数据库 classDatabaseSession:def__init__(self):self.idDB_SESSION_001defexecute(self,query:str):returnf[{self.id}] 执行:{query}defclose(self):print(f[{self.id}] 数据库连接已关闭)defget_db()-Generator[DatabaseSession,None,None]:dbDatabaseSession()try:yielddbfinally:db.close()# 2. API Key 验证 defverify_token(x_token:strHeader(...)):ifx_token!valid-token-123:raiseHTTPException(status_code401,detail无效的 Token)returnx_token# 3. 获取当前用户依赖 tokendefget_current_user(token:strDepends(verify_token)):return{username:alice,token:token}# 路由 app.get(/health)defhealth_check():return{status:OK}app.get(/items/,dependencies[Depends(verify_token)])deflist_items(dbDepends(get_db)):datadb.execute(SELECT name FROM items)return{items:[data]}app.get(/profile)defprofile(userDepends(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组合依赖 —— 验证角色依赖1get_current_user()返回{username: bob, role: user}依赖2require_admin(userDepends(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(userDepends(get_current_user)):ifuser[role]!admin:raiseHTTPException(status_code403,detail需要管理员权限)returnuserapp.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:int0,limit:int10):return{skip:skip,limit:limit}app.get(/products/)deflist_products(paginationDepends(get_pagination)):return{message:产品列表,pagination:pagination}测试/products/→{skip:0, limit:10}/products/?skip20limit5→{skip:20, limit:5}十、小结在本部分你学会了使用Depends()注入依赖函数实现API 验证、数据库会话管理、角色权限控制使用yield实现资源自动清理构建嵌套依赖子依赖在 Swagger 中测试带 Header 的接口