FastAPI 依赖注入1. 核心概念依赖注入Dependency Injection, DI是 FastAPI 的核心设计模式用于复用逻辑— 数据库连接、认证等共享逻辑权限控制— 统一鉴权减少重复— 避免在多个路由中写相同代码2. 基本用法Depends函数作为依赖fromfastapiimportFastAPI,Depends appFastAPI()defcommon_params(q:strNone,skip:int0,limit:int100):return{q:q,skip:skip,limit:limit}app.get(/items/)defread_items(commons:dictDepends(common_params)):returncommonsapp.get(/users/)defread_users(commons:dictDepends(common_params)):returncommonsDepends(common_params)会自动执行函数并将返回值注入。类作为依赖classCommonQueryParams:def__init__(self,q:strNone,skip:int0,limit:int100):self.qq self.skipskip self.limitlimitapp.get(/items/)defread_items(commons:CommonQueryParamsDepends(CommonQueryParams)):return{q:commons.q,skip:commons.skip,limit:commons.limit}简写形式类型声明即依赖app.get(/items/)defread_items(commons:CommonQueryParamsDepends()):return{q:commons.q,skip:commons.skip,limit:commons.limit}3. 子依赖依赖链依赖可以自身也依赖其他依赖defquery_extractor(q:strNone):returnqdefquery_checker(q:strDepends(query_extractor)):ifqandbadinq:raiseHTTPException(status_code400,detailBad query)returnqapp.get(/items/)defread_items(q:strDepends(query_checker)):return{q:q}调用链query_extractor→query_checker→read_items4. 全局依赖应用级依赖defverify_token(token:strHeader(...)):iftoken!secret:raiseHTTPException(status_code401,detailInvalid token)appFastAPI(dependencies[Depends(verify_token)])app.get(/items/)defread_items():return{items:[]}# 自动经过 verify_token 校验app.get(/users/)defread_users():return{users:[]}# 同样自动校验路由级依赖routerAPIRouter(dependencies[Depends(verify_token)])router.get(/items/)defread_items():return{items:[]}单个路由依赖app.get(/admin/,dependencies[Depends(verify_token),Depends(verify_admin)])defadmin_panel():return{message:Welcome admin}5. 实战场景数据库会话fromsqlalchemy.ormimportSessiondefget_db():dbSessionLocal()try:yielddbfinally:db.close()app.get(/books/)deflist_books(db:SessionDepends(get_db)):booksdb.query(Book).all()returnbooksapp.post(/books/)defcreate_book(book:BookCreate,db:SessionDepends(get_db)):db_bookBook(**book.dict())db.add(db_book)db.commit()db.refresh(db_book)returndb_book关键使用yield确保资源释放类似try/finally。认证与权限fromfastapi.securityimportOAuth2PasswordBearerfromjoseimportjwt oauth2_schemeOAuth2PasswordBearer(tokenUrltoken)defget_current_user(token:strDepends(oauth2_scheme),db:SessionDepends(get_db)):payloadjwt.decode(token,SECRET_KEY,algorithms[ALGORITHM])userdb.query(User).filter(User.idpayload[sub]).first()ifnotuser:raiseHTTPException(status_code401,detailInvalid token)returnuserdefget_current_active_user(user:UserDepends(get_current_user)):ifnotuser.is_active:raiseHTTPException(status_code400,detailInactive user)returnuser# 需要登录app.get(/me/)defread_current_user(user:UserDepends(get_current_active_user)):returnuser# 需要管理员权限defget_admin_user(user:UserDepends(get_current_active_user)):ifnotuser.is_admin:raiseHTTPException(status_code403,detailNot enough privileges)returnuserapp.get(/admin/,dependencies[Depends(get_admin_user)])defadmin_panel():return{message:Admin panel}分页依赖classPaginationParams:def__init__(self,page:int1,size:int20,size_max:int100):self.pagemax(1,page)self.sizemin(max(1,size),size_max)self.offset(self.page-1)*self.sizeapp.get(/books/)deflist_books(pagination:PaginationParamsDepends(),db:SessionDepends(get_db)):booksdb.query(Book).offset(pagination.offset).limit(pagination.size).all()return{page:pagination.page,size:pagination.size,data:books}配置与设置frompydanticimportBaseSettingsclassSettings(BaseSettings):database_url:strsqlite:///./test.dbsecret_key:strdev-secretdebug:boolFalseclassConfig:env_file.envdefget_settings():returnSettings()app.get(/info/)defapp_info(settings:SettingsDepends(get_settings)):return{debug:settings.debug}6. yield 依赖资源管理# 数据库连接defget_db():dbSessionLocal()try:yielddbfinally:db.close()# Redis 连接asyncdefget_redis():redisawaitaioredis.from_url(redis://localhost)try:yieldredisfinally:awaitredis.close()app.get(/cache/)asyncdefread_cache(redisDepends(get_redis)):valueawaitredis.get(key)return{value:value}yield 后的清理逻辑defget_db():dbSessionLocal()try:yielddbfinally:db.close()# 请求完成后执行# 也可以用于日志记录deflog_request(request:Request):starttime.time()yielddurationtime.time()-startprint(fRequest took{duration:.2f}s)appFastAPI(dependencies[Depends(log_request)])7. 依赖覆盖测试用defget_db():dbSessionLocal()try:yielddbfinally:db.close()defget_test_db():dbTestSessionLocal()try:yielddbfinally:db.close()appFastAPI()# 测试时覆盖app.dependency_overrides[get_db]get_test_db# 测试结束清除# app.dependency_overrides.clear()8. 依赖缓存同一请求中同一依赖只执行一次结果被缓存复用defexpensive_dependency():print(执行了一次)# 同一请求中只打印一次return{data:expensive}app.get(/test/)deftest(d1:dictDepends(expensive_dependency),d2:dictDepends(expensive_dependency)):# d1 和 d2 是同一个对象return{d1:d1,d2:d2}禁用缓存每次调用都重新执行app.get(/test/)deftest(d1:dictDepends(expensive_dependency,use_cacheFalse)):returnd19. 依赖总结场景方式示例共享查询参数函数依赖Depends(common_params)数据库会话yield 依赖Depends(get_db)认证鉴权依赖链Depends(get_current_user)全局校验应用/路由级依赖dependencies[Depends(...)]测试替换依赖覆盖app.dependency_overrides[fn] mock_fn分页/配置类依赖Depends(PaginationParams)总结FastAPI 的依赖注入系统通过Depends()实现松耦合支持函数/类依赖、依赖链、yield 资源管理、全局依赖和测试覆盖。合理使用可大幅减少重复代码提升可维护性和可测试性。