当前位置：首页 > news >正文

FastAPI-依赖注入

news 2026/3/27 3:28:37

一、什么是依赖注入

依赖注入（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_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid 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_code=404, detail="User 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_cache=False) ): return {"result": result}

查看全文

http://www.jsqmd.com/news/534549/