依賴注入常用于以下場景:
-
共享業務邏輯(復用相同的代碼邏輯)
-
共享數據庫連接
-
實現安全、驗證、角色權限
-
等……
上述場景均可以使用依賴注入,將代碼重復最小化。
創建依賴項
依賴項就是一個函數,且可以使用與路徑操作函數相同的參數,依賴項函數的形式和結構與路徑操作函數一樣。因此,可以把依賴項當作沒有「裝飾器」(即,沒有 @app.get("/some-path") )的路徑操作函數。
from typing import Unionimport uvicorn
from fastapi import FastAPI, Dependsapp = FastAPI()# 依賴的組件函數
async def common_parameters(q: Union[str, None] = None,skip: int = 0,limit: int = 100,
):return {"skip": skip, "limit": limit, "q": q}@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 commonsif __name__ == '__main__':uvicorn.run(app)
實現過程:
1、導入 Depends
from fastapi import Depends
2、定義依賴的組件函數
# 依賴的組件函數
async def common_parameters(q: Union[str, None] = None,skip: int = 0,limit: int = 100,
):return {"skip": skip, "limit": limit, "q": q}
3、聲明依賴項
聲明依賴項需要使用 Depends 和一個新的參數:
async def read_items(commons: dict = Depends(common_parameters)):
其他與「依賴注入」概念相同的術語為:
-
資源(Resource)
-
提供方(Provider)
-
服務(Service)
-
可注入(Injectable)
-
組件(Component)
依賴注入系統如此簡潔的特性,讓 FastAPI 可以與下列系統兼容:
-
關系型數據庫
-
NoSQL 數據庫
-
外部支持庫
-
外部 API
-
認證和鑒權系統
-
API 使用監控系統
-
響應數據注入系統
-
等等……
類作為依賴項
依賴項時一個可調用的對象,Python 中的 "可調用對象" 是指任何 Python 可以像函數一樣 "調用" 的對象。Python 類也是 可調用對象。因此,在 FastAPI 中,你可以使用一個 Python 類作為一個依賴項。
實際上 FastAPI 檢查的是它是一個 "可調用對象"(函數,類或其他任何類型)以及定義的參數。
import uvicorn
from fastapi import FastAPI, Dependsapp = FastAPI()fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]class CommonQueryParams:def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):self.q = qself.skip = skipself.limit = limit# FastAPI 調用 CommonQueryParams 類。這將創建該類的一個 "實例",該實例將作為參數 commons 被傳遞給你的函數。
@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):response = {}if commons.q:response.update({"q": commons.q})items = fake_items_db[commons.skip: commons.skip + commons.limit]response.update({"items": items})return responseif __name__ == '__main__':uvicorn.run(app, host="127.0.0.1", port=8000)
FastAPI 為這些情況提供了一個快捷方式,在這些情況下,依賴項 明確地 是一個類,FastAPI 將 "調用" 它來創建類本身的一個實例。
快捷寫法
commons: CommonQueryParams = Depends()
分層依賴注入
假設是用戶管理模塊,創建如下的目錄結構以及python文件:
schemas/user_schema.py
from pydantic import BaseModel, Field
from datetime import datetimeclass User(BaseModel):id: int | None = Field(description="主鍵ID")username: str | None = Field(description="用戶名")age: int | None = Field(description="年齡")is_vip: int | None = Field(description="是否會員 0-不是,1-是")create_time: datetime = Field(description="創建時間")
dao/user_dao.py
from datetime import datetime
from user.schemas.user_schema import Userclass UserDAO:"""數據層"""def __init__(self):passdef get_user_by_username(self, username)-> User:"""根據用戶名查詢用戶信息:param username::return:"""user = User(id=1, username="alice", age=25, is_vip=0, create_time=datetime.now())return user
service/user_service.py
from fastapi import Dependsfrom user.dao.user_dao import UserDAO
from user.schemas.user_schema import Userclass UserService:"""業務邏輯處理層"""def __init__(self, user_dao: UserDAO = Depends()):self.user_dao = user_daodef find_by_username(self, username: str)-> User:"""根據用戶名查詢用戶信息:param username::return:"""return self.user_dao.get_user_by_username(username)
routers/user_routers.py
from fastapi import APIRouter, Dependsfrom commons.result import Result
from user.exceptions.user_exception import UserNotFoundException
from user.service.user_serivce import UserServiceusers = APIRouter()@users.get(path="/find/{username}", summary="根據用戶名查詢用戶", response_model=Result)
async def find_user_by_username(username: str, user_service: UserService = Depends()):user = user_service.find_by_username(username)# 用戶不存在if user is None:raise UserNotFoundExceptionresult = Result()result.data = userreturn result
commoms/result.py
from typing import Anyfrom pydantic import BaseModel, Fieldclass Result(BaseModel):code: int = Field(description="響應狀態碼", default=200)message: str = Field(description="響應結果信息", default="成功")data: Any = Field(description="響應結果數據", default=None)