依赖注入-Depends

常用场景：

  1. 共享业务逻辑
  2. 共享数据库链接
  3. 实现安全、验证、角色权限等
  • 导入Depends,传参为可调用对象
  2. """依赖注入-Depends"""
  3. async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
  4.     return {"q": q, 'skip': skip, "limit": limit}
  6. @app.get("/items17")
  7. async def read_depends(commons: dict= Depends(common_parameters)):
  8.     """依赖注入-Depends"""
  9.     return commons

类依赖注入

  1. class CommonQueryParams:
  2.     def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
  3.         self.q = q
  4.         self.skip = skip
  5.         self.limit = limit
  7. @app.post("/items18")
  8. async def read_items18(commons: CommonQueryParams = Depends()):
  9.     if commons.q is not None:
  10.         logger.info(f"q不为None")
  11.     return commons

路径操作依赖项-dependencies

  1. 路径操作函数中使用的依赖项不需要返回值，例如信息验证、token等等，可在路径餐做函数中添加可选参数dependencies
  2. dependencies参数为Depends()组成的list
  2. """路径操作依赖-dependencies"""
  3. async def verify_token(x_token: str = Header(...)):
  4.     if x_token != "fake-super-secret-token":
  5.         raise HTTPException(status_code=400, detail="X-Token header invalid")
  8. async def verify_key(x_key: str = Header(...)):
  9.     if x_key != "fake-super-secret-key":
  10.         raise HTTPException(status_code=400, detail="X-Key header invalid")
  11.     return x_key
  14. @app.get("/items18/", dependencies=[Depends(verify_token), Depends(verify_key)])
  15. async def read_items18():
  16.     return [{"item": "Foo"}, {"item": "Bar"}]

全局依赖

app 实例创建时添加dependencies参数注入依赖
app = FastAPI(dependencies=[Depends(…), …])

  1. """全局依赖"""
  2. async def verify_token(x_token: str = Header(...)):
  3.     if x_token != "fake-super-secret-token":
  4.         raise HTTPException(status_code=400, detail="X-Token header invalid")
  7. async def verify_key(x_key: str = Header(...)):
  8.     if x_key != "fake-super-secret-key":
  9.         raise HTTPException(status_code=400, detail="X-Key header invalid")
  10.     return x_key
  13. app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])
  16. @app.get("/items/")
  17. async def read_items():
  18.     return [{"item": "Portal Gun"}, {"item": "Plumbus"}]