FastAPI 依赖注入

FastAPI 依赖注入

1. 核心概念

依赖注入（Dependency Injection, DI）是 FastAPI 的核心设计模式，用于：

• 复用逻辑— 数据库连接、认证等共享逻辑
• 权限控制— 统一鉴权
• 减少重复— 避免在多个路由中写相同代码

2. 基本用法（Depends）

函数作为依赖

fromfastapiimportFastAPI,Depends app=FastAPI()defcommon_params(q:str=None,skip:int=0,limit:int=100):return{"q":q,"skip":skip,"limit":limit}@app.get("/items/")defread_items(commons:dict=Depends(common_params)):returncommons@app.get("/users/")defread_users(commons:dict=Depends(common_params)):returncommons

Depends(common_params)会自动执行函数并将返回值注入。

类作为依赖

classCommonQueryParams:def__init__(self,q:str=None,skip:int=0,limit:int=100):self.q=q self.skip=skip self.limit=limit@app.get("/items/")defread_items(commons:CommonQueryParams=Depends(CommonQueryParams)):return{"q":commons.q,"skip":commons.skip,"limit":commons.limit}

简写形式（类型声明即依赖）：

@app.get("/items/")defread_items(commons:CommonQueryParams=Depends()):return{"q":commons.q,"skip":commons.skip,"limit":commons.limit}

3. 子依赖（依赖链）

依赖可以自身也依赖其他依赖：

defquery_extractor(q:str=None):returnqdefquery_checker(q:str=Depends(query_extractor)):ifqand"bad"inq:raiseHTTPException(status_code=400,detail="Bad query")returnq@app.get("/items/")defread_items(q:str=Depends(query_checker)):return{"q":q}

调用链：query_extractor→query_checker→read_items

4. 全局依赖

应用级依赖

defverify_token(token:str=Header(...)):iftoken!="secret":raiseHTTPException(status_code=401,detail="Invalid token")app=FastAPI(dependencies=[Depends(verify_token)])@app.get("/items/")defread_items():return{"items":[]}# 自动经过 verify_token 校验@app.get("/users/")defread_users():return{"users":[]}# 同样自动校验

路由级依赖

router=APIRouter(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():db=SessionLocal()try:yielddbfinally:db.close()@app.get("/books/")deflist_books(db:Session=Depends(get_db)):books=db.query(Book).all()returnbooks@app.post("/books/")defcreate_book(book:BookCreate,db:Session=Depends(get_db)):db_book=Book(**book.dict())db.add(db_book)db.commit()db.refresh(db_book)returndb_book

关键：使用yield确保资源释放（类似try/finally）。

认证与权限

fromfastapi.securityimportOAuth2PasswordBearerfromjoseimportjwt oauth2_scheme=OAuth2PasswordBearer(tokenUrl="token")defget_current_user(token:str=Depends(oauth2_scheme),db:Session=Depends(get_db)):payload=jwt.decode(token,SECRET_KEY,algorithms=[ALGORITHM])user=db.query(User).filter(User.id==payload["sub"]).first()ifnotuser:raiseHTTPException(status_code=401,detail="Invalid token")returnuserdefget_current_active_user(user:User=Depends(get_current_user)):ifnotuser.is_active:raiseHTTPException(status_code=400,detail="Inactive user")returnuser# 需要登录@app.get("/me/")defread_current_user(user:User=Depends(get_current_active_user)):returnuser# 需要管理员权限defget_admin_user(user:User=Depends(get_current_active_user)):ifnotuser.is_admin:raiseHTTPException(status_code=403,detail="Not enough privileges")returnuser@app.get("/admin/",dependencies=[Depends(get_admin_user)])defadmin_panel():return{"message":"Admin panel"}

分页依赖

classPaginationParams:def__init__(self,page:int=1,size:int=20,size_max:int=100):self.page=max(1,page)self.size=min(max(1,size),size_max)self.offset=(self.page-1)*self.size@app.get("/books/")deflist_books(pagination:PaginationParams=Depends(),db:Session=Depends(get_db)):books=db.query(Book).offset(pagination.offset).limit(pagination.size).all()return{"page":pagination.page,"size":pagination.size,"data":books}

配置与设置

frompydanticimportBaseSettingsclassSettings(BaseSettings):database_url:str="sqlite:///./test.db"secret_key:str="dev-secret"debug:bool=FalseclassConfig:env_file=".env"defget_settings():returnSettings()@app.get("/info/")defapp_info(settings:Settings=Depends(get_settings)):return{"debug":settings.debug}

6. yield 依赖（资源管理）

# 数据库连接defget_db():db=SessionLocal()try:yielddbfinally:db.close()# Redis 连接asyncdefget_redis():redis=awaitaioredis.from_url("redis://localhost")try:yieldredisfinally:awaitredis.close()@app.get("/cache/")asyncdefread_cache(redis=Depends(get_redis)):value=awaitredis.get("key")return{"value":value}

yield 后的清理逻辑

defget_db():db=SessionLocal()try:yielddbfinally:db.close()# 请求完成后执行# 也可以用于日志记录deflog_request(request:Request):start=time.time()yieldduration=time.time()-startprint(f"Request took{duration:.2f}s")app=FastAPI(dependencies=[Depends(log_request)])

7. 依赖覆盖（测试用）

defget_db():db=SessionLocal()try:yielddbfinally:db.close()defget_test_db():db=TestSessionLocal()try:yielddbfinally:db.close()app=FastAPI()# 测试时覆盖app.dependency_overrides[get_db]=get_test_db# 测试结束清除# app.dependency_overrides.clear()

8. 依赖缓存

同一请求中，同一依赖只执行一次，结果被缓存复用：

defexpensive_dependency():print("执行了一次")# 同一请求中只打印一次return{"data":"expensive"}@app.get("/test/")deftest(d1:dict=Depends(expensive_dependency),d2:dict=Depends(expensive_dependency)):# d1 和 d2 是同一个对象return{"d1":d1,"d2":d2}

禁用缓存（每次调用都重新执行）：

@app.get("/test/")deftest(d1:dict=Depends(expensive_dependency,use_cache=False)):returnd1

9. 依赖总结

总结：FastAPI 的依赖注入系统通过Depends()实现松耦合，支持函数/类依赖、依赖链、yield 资源管理、全局依赖和测试覆盖。合理使用可大幅减少重复代码，提升可维护性和可测试性。