FastAPI-依赖注入

一、什么是依赖注入依赖注入Dependency Injection是一种设计模式用于管理组件之间的依赖关系。在 FastAPI 中它用于共享数据库连接强制执行安全性和认证参数验证代码复用二、基础依赖from fastapi import Depends, FastAPI app FastAPI() # 简单依赖函数 def common_parameters(q: str None, skip: int 0, limit: int 100): return {q: q, skip: skip, limit: limit} app.get(/items/) async def read_items(commons: dict Depends(common_parameters)): return commons app.get(/users/) async def read_users(commons: dict Depends(common_parameters)): return commons三、类作为依赖from fastapi import Depends class CommonQueryParams: def __init__(self, q: str None, skip: int 0, limit: int 100): self.q q self.skip skip self.limit limit app.get(/items/) async def read_items(commons: CommonQueryParams Depends()): # 自动推断依赖类型 return { q: commons.q, skip: commons.skip, limit: commons.limit }四、多层依赖from fastapi import Depends, HTTPException, status # 第一层获取 token def get_token(token: str Header()): return token # 第二层验证 token依赖第一层 def verify_token(token: str Depends(get_token)): if token ! secret-token: raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detailInvalid token ) return token # 第三层获取当前用户依赖第二层 async def get_current_user( token: str Depends(verify_token), db: AsyncSession Depends(get_db) ): # 根据 token 查询用户 user await get_user_by_token(db, token) if not user: raise HTTPException(status_code404, detailUser not found) return user # 路由使用最终依赖 app.get(/users/me) async def read_current_user(current_user Depends(get_current_user)): return current_user五、 全局依赖# 应用级依赖所有路由都会执行 app FastAPI(dependencies[Depends(verify_token)]) # 路由级依赖某个路由组的所有端点 router APIRouter( prefix/admin, dependencies[Depends(verify_admin)] )六、yield 依赖async def get_db(): db AsyncSessionLocal() try: yield db # 使用 yield 而非 return finally: await db.close() # 请求完成后自动清理 app.get(/users/) async def get_users(db: AsyncSession Depends(get_db)): # db 使用完毕后会自动关闭 return await get_all_users(db)七、依赖缓存from fastapi import Depends # 默认情况下同一个请求中多次调用同一依赖会被缓存 async def expensive_operation(): # 昂贵的计算 return result app.get(/test) async def test_endpoint( # 两次调用但只执行一次 result1 Depends(expensive_operation), result2 Depends(expensive_operation) ): return {result1: result1, result2: result2} # 禁用缓存 app.get(/test2) async def test_endpoint2( result Depends(expensive_operation, use_cacheFalse) ): return {result: result}