跳至內容

依賴項

FastAPI 有一個非常強大但直觀的 依賴注入 系統。

它被設計成非常易於使用,並且讓任何開發者都能輕鬆地將其他組件與 FastAPI 整合。

什麼是「依賴注入」

在程式設計中,「依賴注入」 意味著您的程式碼(在這種情況下,您的路徑操作函式)有一種方法可以宣告它需要工作和使用的東西:「依賴項」。

然後,該系統(在這種情況下是 FastAPI)將負責執行所有必要的操作,為您的程式碼提供所需的依賴項(「注入」依賴項)。

這在您需要以下情況時非常有用:

  • 具有共享邏輯(相同的程式碼邏輯重複使用)。
  • 共享資料庫連線。
  • 強制執行安全性、驗證、角色要求等。
  • 以及許多其他事情……

所有這些,同時最大限度地減少程式碼重複。

第一步

讓我們看一個非常簡單的例子。目前它非常簡單,以至於沒有什麼用處。

但這樣我們就可以專注於 依賴注入 系統是如何工作的。

建立依賴項,或「可依賴項」

讓我們先關注依賴項。

它只是一個函式,可以接受所有與路徑操作函式相同的參數

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Annotated, Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

提示

如果可能,盡量使用 Annotated 版本。

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = 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

提示

如果可能,盡量使用 Annotated 版本。

from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = 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

就這樣。

2 行.

它與所有路徑操作函式具有相同的形狀和結構。

您可以將其視為沒有「裝飾器」(沒有 @app.get("/some-path"))的路徑操作函式

它可以返回您想要的任何內容。

在這種情況下,此依賴項預期

  • 一個可選的查詢參數 q,它是一個 str
  • 一個可選的查詢參數 skip,它是一個 int,預設值為 0
  • 一個可選的查詢參數 limit,它是一個 int,預設值為 100

然後它只返回一個包含這些值的 dict

資訊

FastAPI 在 0.95.0 版中增加了對 Annotated 的支援(並開始推薦使用它)。

如果您使用的是舊版本,則在嘗試使用 Annotated 時會出現錯誤。

在使用 Annotated 之前,請確保您升級 FastAPI 版本至至少 0.95.1。

導入 Depends

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Annotated, Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

提示

如果可能,盡量使用 Annotated 版本。

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = 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

提示

如果可能,盡量使用 Annotated 版本。

from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = 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

在「依賴對象」中宣告依賴項

與您在路徑操作函式參數中使用 BodyQuery 等方式相同,在新參數中使用 Depends

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Annotated, Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons
from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons


@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

提示

如果可能,盡量使用 Annotated 版本。

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = 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

提示

如果可能,盡量使用 Annotated 版本。

from typing import Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = 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

雖然您在函式參數中使用 Depends 的方式與使用 BodyQuery 等相同,但 Depends 的工作方式略有不同。

您只給 Depends 一個參數。

此參數必須類似於函式。

您**不需要直接呼叫它**(不要在結尾加上括號),只需將其作為參數傳遞給 Depends() 即可。

這個函式的參數傳遞方式與*路徑操作函式*相同。

提示

在下一章中,您將看到除了函式之外,還有哪些「東西」可以用作依賴項。

每當有新的請求到達時,**FastAPI** 會負責:

  • 使用正確的參數呼叫您的依賴項(「可依賴」)函式。
  • 取得函式的結果。
  • 將該結果指派給*路徑操作函式*中的參數。
graph TB

common_parameters(["common_parameters"])
read_items["/items/"]
read_users["/users/"]

common_parameters --> read_items
common_parameters --> read_users

透過這種方式,您只需編寫一次共用程式碼,**FastAPI** 就會負責為您的*路徑操作*呼叫它。

請注意

您不需要建立一個特殊的類別並將其傳遞給 **FastAPI** 來「註冊」它或執行任何類似的操作。

您只需將其傳遞給 Depends,**FastAPI** 就知道如何處理其餘部分。

共用 Annotated 依賴項

在上面的範例中,您會看到有一點點的**程式碼重複**。

當您需要使用 common_parameters() 依賴項時,您必須使用型別註釋和 Depends() 完整地撰寫參數。

commons: Annotated[dict, Depends(common_parameters)]

但因為我們使用的是 Annotated,我們可以將該 Annotated 值儲存在一個變數中,並在多個地方使用它。

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons
from typing import Annotated, Union

from fastapi import Depends, FastAPI

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons
from typing import Union

from fastapi import Depends, FastAPI
from typing_extensions import Annotated

app = FastAPI()


async def common_parameters(
    q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
    return {"q": q, "skip": skip, "limit": limit}


CommonsDep = Annotated[dict, Depends(common_parameters)]


@app.get("/items/")
async def read_items(commons: CommonsDep):
    return commons


@app.get("/users/")
async def read_users(commons: CommonsDep):
    return commons

提示

這只是標準的 Python 語法,稱為「型別別名」,它實際上並不是 **FastAPI** 特有的。

但由於 **FastAPI** 基於 Python 標準,包含 Annotated,您可以在您的程式碼中使用這個技巧。😎

依賴項將會繼續如預期般運作,**最棒的是型別資訊將會被保留**,這表示您的編輯器將能夠繼續提供**自動完成**、**行內錯誤**等功能。其他工具如 mypy 也是如此。

當您在**大型程式碼庫**中,在**許多*路徑操作***中重複使用**相同的依賴項**時,這將特別有用。

使用 async 或不使用 async

由於依賴項也將由 **FastAPI** 呼叫(與您的*路徑操作函式*相同),因此在定義函式時,相同的規則也適用。

您可以使用 async def 或一般的 def

您可以在一般的 def *路徑操作函式*中宣告使用 async def 的依賴項,或在 async def *路徑操作函式*中宣告使用 def 的依賴項,依此類推。

這沒有關係。**FastAPI** 會知道該怎麼做。

備註

如果您不清楚,請查看文件中關於 asyncawait非同步:「趕時間嗎?」章節。

與 OpenAPI 整合

您的依賴項(和子依賴項)的所有請求宣告、驗證和需求都將整合到相同的 OpenAPI 模式中。

因此,互動式文件也將包含來自這些依賴項的所有資訊。

簡單用法

仔細觀察,*路徑操作函式*被宣告為在*路徑*和*操作*匹配時使用,然後 **FastAPI** 負責使用正確的參數呼叫函式,從請求中提取資料。

實際上,所有(或大多數)Web 框架都以相同的方式運作。

您永遠不會直接呼叫這些函式。它們是由您的框架(在這個例子中是 FastAPI)呼叫的。

使用依賴注入系統,您也可以告訴 FastAPI 您的*路徑操作函式*也「依賴」於其他應該在您的*路徑操作函式*之前執行的東西,而 FastAPI 將會負責執行它並「注入」結果。

其他表示相同「依賴注入」概念的常見術語有:

  • 資源 (resources)
  • 提供者 (providers)
  • 服務 (services)
  • 可注入物件 (injectables)
  • 組件 (components)

FastAPI 插件

整合和「插件」可以使用**依賴注入**系統構建。但事實上,實際上**不需要建立「插件」**,因為通過使用依賴項,可以宣告無限數量的整合和互動,這些整合和互動可供您的*路徑操作函式*使用。

而且,依賴項的建立方式非常簡單直觀,讓您可以只導入所需的 Python 套件,並將它們與您的 API 函式整合,* буквально 只需幾行程式碼*。

您將在接下來關於關聯式和 NoSQL 資料庫、安全性等的章節中看到相關範例。

FastAPI 相容性

依賴注入系統的簡潔性使 FastAPI 與以下各項相容:

  • 所有關聯式資料庫
  • NoSQL 資料庫
  • 外部套件
  • 外部 API
  • 驗證和授權系統
  • API 使用情況監控系統
  • 回應資料注入系統
  • 等等

簡單而強大

雖然階層式依賴注入系統的定義和使用非常簡單,但它仍然非常強大。

您可以定義依賴項,而這些依賴項本身也可以定義依賴項。

最終,會構建一個階層式的依賴樹,而**依賴注入**系統會負責為您解決所有這些依賴項(及其子依賴項),並在每個步驟提供(注入)結果。

例如,假設您有 4 個 API 端點(*路徑操作*)

  • /items/public/
  • /items/private/
  • /users/{user_id}/activate
  • /items/pro/

那麼您只需使用依賴項和子依賴項,即可為每個端點新增不同的權限要求。

graph TB

current_user(["current_user"])
active_user(["active_user"])
admin_user(["admin_user"])
paying_user(["paying_user"])

public["/items/public/"]
private["/items/private/"]
activate_user["/users/{user_id}/activate"]
pro_items["/items/pro/"]

current_user --> active_user
active_user --> admin_user
active_user --> paying_user

current_user --> public
active_user --> private
admin_user --> activate_user
paying_user --> pro_items

OpenAPI 整合

所有這些依賴項在宣告其需求的同時,也會將參數、驗證等新增到您的*路徑操作*中。

FastAPI 將負責將所有這些新增到 OpenAPI 結構描述中,以便在互動式文件系統中顯示。