學習!FastAPI

目錄

  • FastAPI簡介
  • 快速開始
    • 安裝FastApi
    • FastAPI CLI
    • 自動化文檔
  • Reqeust
    • 路徑參數
      • Enum 類用于路徑參數
      • 路徑參數和數值校驗
    • 查詢參數
      • 查詢參數和字符串校驗
    • 請求體
      • 多個請求體參數
      • 嵌入單個請求體參數
    • Cookie
    • Header
    • 表單
    • 文件
    • 直接使用請求
  • Response
    • Response Model
      • 多個關聯模型
    • 響應狀態碼
    • 處理錯誤
      • 自定義異常處理
    • 直接返回響應
      • HTML,流,文件和其他
    • Cookies
    • 響應頭
  • 路徑操作
    • 路徑操作配置
    • 依賴項
      • 類作為依賴項
      • 子依賴項
      • 路徑操作裝飾器依賴項
      • 全局依賴項
      • 使用yield的依賴項
    • 高級依賴項
    • 子路由APIRouter
    • 為什么這樣配置?
  • 中間件
    • CORS(跨域資源共享)
    • 其他中間件
    • 和依賴項的區別
  • 后臺任務
  • 生命周期事件
  • SQL(關系型)數據庫——現代化ORM
  • 靜態文件
    • 模板
  • 部署

FastAPI簡介

FastAPI 是一個現代、快速(高性能)的 Web 框架,用于構建基于 Python 的 API。它是一個開源項目,基于 Starlette 和 Pydantic 庫構建而成,提供了強大的功能和高效的性能 。其核心特征如下:

  • 高性能:FastAPI 基于異步編程模型(使用 Python 的 asyncawait 關鍵字),利用了 Python 的異步生態系統,提供出色的性能和吞吐量 。
  • 自動文檔生成:FastAPI 可以自動為你的 API 生成交互式文檔,支持自動檢測請求參數和響應模型,并生成相應的 API 文檔 。
  • 數據驗證和轉換:FastAPI 使用 Pydantic 庫,提供了強大的數據驗證和轉換功能,能夠自動處理請求和響應數據的驗證、轉換和序列化 。
  • 類型提示:FastAPI 基于 Python 的類型提示機制,提供了強類型的請求和響應處理,這樣可以減少很多常見的錯誤,并提供更好的代碼提示和可讀性 。
  • 安全認證:FastAPI 支持常用的認證方式,如 OAuth2、JWT 等,并提供了對 HTTPS 的支持,可以保護你的 API 通信安全 。
  • 強大的生態系統:FastAPI 可以與眾多 Python 生態系統中的工具和庫無縫集成,如 SQLAlchemy、Databases、Redis 等 。

與 Django 和 Flask 的對比

特性DjangoFlaskFastAPI
框架類型全功能框架,包含 ORM、模板引擎等微框架,靈活輕量微框架,專注于構建高性能 API
異步支持原生不支持異步,需要額外配置原生不支持異步,需要額外配置原生支持異步編程
類型提示支持支持有限,主要依賴第三方庫支持有限,主要依賴第三方庫原生支持,利用 Python 的類型提示
自動文檔生成需要手動配置或使用第三方庫需要手動配置或使用第三方庫自動生成,支持 Swagger UI 和 ReDoc
性能較低,適合中小型項目中等,適合小型項目高性能,適合構建高并發、高性能的 API 服務
學習曲線陡峭,需要學習完整的框架結構平緩,適合初學者平緩,適合有一定 Python 基礎的開發者
社區支持成熟,擁有龐大的社區和豐富的資源成熟,擁有活躍的社區和豐富的插件新興,社區在快速增長,資源逐漸豐富

為什么學習 FastAPI 是必要的?

  1. 現代 Web 開發的需求:隨著前后端分離架構的流行,構建高性能的 RESTful API 成為主流需求。FastAPI 專注于 API 的構建,滿足現代 Web 開發的需求。

  2. 高性能的優勢:FastAPI 的異步支持和高性能特性,使其在處理高并發請求時表現出色,適合構建需要高吞吐量的應用,如實時數據處理、機器學習模型部署等。

  3. 提升開發效率:自動文檔生成、類型提示支持和數據驗證功能,減少了重復性工作,提高了開發效率和代碼質量。

  4. 良好的可維護性:強類型支持和清晰的代碼結構,使得項目更易于維護和擴展,適合長期發展的項目。

  5. 快速增長的社區:雖然 FastAPI 是一個相對較新的框架,但其社區在快速增長,越來越多的企業和開發者開始采用 FastAPI,學習它有助于跟上技術發展的步伐。

最后在部署 Python Web 應用時,選擇合適的服務器對性能和擴展性至關重要。以下是對 Flask 和 FastAPI 在部署方面的對比,特別關注 Gunicorn、uWSGI 和 Uvicorn 的性能表現。

  • Gunicorn
  • 優點:成熟穩定,配置簡單,適合中等負載的應用。
  • 缺點:在高并發場景下,性能可能下降,延遲增加。
  • uWSGI
  • 優點:功能豐富,支持多線程,適合處理大量請求。
  • 缺點:配置復雜,調優難度較大。

實測數據顯示,Flask 在使用 Gunicorn 或 uWSGI 部署時的吞吐量和延遲表現不如異步框架。

FastAPI 是基于 ASGI 的現代異步框架,常與 Uvicorn 或 Gunicorn(結合 UvicornWorker)部署,具備更高的性能和并發處理能力。(scien.cx)

  • Uvicorn
  • 優點:輕量高效,原生支持異步,適合小型或中等規模的應用。
  • 缺點:在大型生產環境中,可能缺乏進程管理等高級功能。(scien.cx)
  • Gunicorn + UvicornWorker
  • 優點:結合了 Gunicorn 的進程管理和 Uvicorn 的異步性能,適合高并發、大規模的生產環境。
  • 缺點:配置相對復雜,需要額外安裝 UvicornWorker。(scien.cx, GitHub)

實測數據顯示,FastAPI 在使用 Gunicorn + UvicornWorker 部署時,吞吐量和響應時間優于 Flask 的部署方式。

框架部署方式吞吐量(req/s)平均響應時間(ms)適用場景
FlaskGunicorn約 3,500較高中小型應用,低并發
FlaskuWSGI略高于 Gunicorn略低于 Gunicorn需要多線程支持的應用
FastAPIUvicorn約 11,000較低中等規模,異步應用
FastAPIGunicorn + UvicornWorker約 20,000更低高并發,大型生產環境(kisspeter.github.io, scien.cx, Gist, GitHub, php.cn, Medium, GitHub)

通過以上數據對比,我們可以直觀的看出FastApi性能之強悍,尤其是現在AI時代,得益于python在機器學習方面的主場優勢,很多AI平臺應用都是基于langChain+FastApi+AI+SSE 去部署,AI的迅猛發展無疑為FastApi注入了新鮮血液。

其次在AI時代,Web開發已經不再局限于以往的條條框框,我們更偏向于將社區的高性能組件自行拼接達到1 x 1 x 1 … x1 >= 1的效果。而非一家獨大,如django框架,當初我學習django的時候最新版本號是3.x,現在已經達到了5.x,django逐漸引入了異步視圖,優化了orm和forms組件… 即便如此forms不如pydantic,orm在現代非結構化數據庫中沒有一絲用武之地。厚重的框架約束了開發者的思想和創造力, 隨即我將目光投向了Flask!

Flask是一個微框架,在python協程還處在feature的歲月,Flask熱火朝天,他的高自由度很快吸引一些中高等水平開發者的眼球,uwsgi/gunicorn + Flask部署的服務很快占到主流(Flask是wsgi框架,如果想要異步視圖,需要先用 asgiref中轉,再用uvicron部署)。 好景不長,python asyncio迅猛發展,anyio,asyncior,uvicorn…等高性能組件如雨后春筍一般現世,終于來到了FastApi的世界!

看到這里,也就剩下了一個念頭,拋棄歷史包袱,學習FastAPI!

快速開始

FastApi官方提供了非常全面的中文文檔:https://fastapi.tiangolo.com/zh/learn/

Starlette (和 FastAPI) 是基于 AnyIO 實現的,這使得它們可以兼容 Python 的標準庫 asyncio 和 Trio。(通過 FastAPI 你可以獲得所有 Starlette 的特性 ( FastAPI 就像加強版的 Starlette ) )

特別是,你可以直接使用 AnyIO 來處理高級的并發用例,這些用例需要在自己的代碼中使用更高級的模式。

即使你沒有使用 FastAPI,你也可以使用 AnyIO 編寫自己的異步程序,使其擁有較高的兼容性并獲得一些好處(例如, 結構化并發)。

我(指原作者)基于 AnyIO 新建了一個庫,作為一個輕量級的封裝層,用來優化類型注解,同時提供了更好的自動補全、內聯錯誤提示等功能。這個庫還附帶了一個友好的入門指南和教程,能幫助你理解并編寫自己的異步代碼:Asyncer。如果你有結合使用異步代碼和常規(阻塞/同步)代碼的需求,這個庫會特別有用。

安裝FastApi

那么現在你就可以通過pip去安裝FastAPI:pip install "fastapi[standard]"

有人這時候比較疑惑,standard是做什么的?

在 Python 的包管理中,方括號 [] 表示安裝額外的可選依賴項。因此,fastapi[standard] 會安裝 FastAPI 的核心功能以及被標記為 standard 的可選依賴項。

因此pip install "fastapi[standard]"pip install fastapi 之間存在顯著區別。前者安裝了 FastAPI 的核心功能以及一組常用的可選依賴項,而后者僅安裝了 FastAPI 的核心功能。(fastapi.org.cn)

根據 FastAPI 的官方文檔,standard 組包含以下額外依賴項:

  • email-validator:用于驗證電子郵件地址的格式。
  • httpx:用于測試客戶端(TestClient)的 HTTP 請求。
  • jinja2:用于模板渲染。
  • python-multipart:用于處理表單數據(request.form())。
  • uvicorn:ASGI 服務器,用于運行 FastAPI 應用。
  • fastapi-cli:提供命令行工具 fastapi。 (pypi.ac.cn, fastapi.org.cn)

這些依賴項使得 FastAPI 更加功能完整,適用于開發和測試階段。(fastapi.org.cn)

  • 僅安裝核心功能:如果您只需要 FastAPI 的基本功能,可以使用:

    pip install fastapi
    
  • 安裝標準依賴項:如果您需要上述提到的額外功能,可以使用:

    pip install "fastapi[standard]"
    

FastAPI CLI

FastAPI CLI 是一個命令行程序,你可以用它來部署和運行你的 FastAPI 應用程序,管理你的 FastAPI 項目,等等。

要在開發環境中運行你的 FastAPI 應用,你可以使用 fastapi dev 命令:

fastapi dev main.py

該命令行程序 fastapi 就是 FastAPI CLI。FastAPI CLI 接收你的 Python 程序路徑,自動檢測包含 FastAPI 的變量(通常命名為 app)及其導入方式,然后啟動服務。

在生產環境中,你應該使用 fastapi run 命令。🚀

在內部,**FastAPI CLI 使用了 Uvicorn,這是一個高性能、適用于生產環境的 ASGI 服務器。**😎

注意:當你運行 fastapi dev 時,它將以開發模式運行。默認情況下,它會啟用自動重載,因此當你更改代碼時,它會自動重新加載服務器。該功能是資源密集型的,且相較不啟用時更不穩定,因此你應該僅在開發環境下使用它。

廢話不多說,我們開始第一行FastApi代碼:

from fastapi import FastAPIapp = FastAPI()@app.get("/")
async def root():return {"message": "Hello World"}

運行實時服務器:fastapi dev main.py

FastAPI   Starting development server 🚀Searching for package file structure from directorieswith __init__.py filesImporting from /home/user/code/awesomeappmodule   🐍 main.pycode   Importing the FastAPI app object from the module withthe following code:from main import appapp   Using import string: main:appserver   Server started at http://127.0.0.1:8000server   Documentation at http://127.0.0.1:8000/docstip   Running in development mode, for production use:fastapi runLogs:INFO   Will watch for changes in these directories:['/home/user/code/awesomeapp']INFO   Uvicorn running on http://127.0.0.1:8000 (Press CTRL+Cto quit)INFO   Started reloader process [383138] using WatchFilesINFO   Started server process [383153]INFO   Waiting for application startup.INFO   Application startup complete.

打開瀏覽器訪問 http://127.0.0.1:8000,你將看到如下的 JSON 響應:

{"message": "Hello World"}

自動化文檔

跳轉到 http://127.0.0.1:8000/docs,你將會看到自動生成的交互式 API 文檔(由 Swagger UI 提供):

在這里插入圖片描述
前往 http://127.0.0.1:8000/redoc,你將會看到可選的自動生成文檔 (由 ReDoc 提供):

在這里插入圖片描述

在 FastAPI 中,術語“交互式 API 文檔”和“可選的 API 文檔”通常指的是兩種不同風格的自動生成的文檔界面:Swagger UIReDoc。這兩者都基于 OpenAPI 規范,提供了交互式的 API 文檔體驗,但在界面設計和功能側重點上有所不同。(DeepWiki, 菜鳥教程)

🔍 交互式 API 文檔(Swagger UI)

  • 訪問地址:默認情況下,運行 FastAPI 應用后,可以通過 http://127.0.0.1:8000/docs 訪問。(菜鳥教程)

  • 主要特點

    • 交互性強:提供“Try it out”按鈕,允許用戶直接在瀏覽器中測試 API 端點。
    • 實時反饋:在文檔中填寫參數并執行請求后,立即顯示響應結果,方便調試。
    • 界面直觀:以卡片形式展示每個端點,結構清晰,便于瀏覽。(菜鳥教程, 博客園)
  • 適用場景:開發階段頻繁測試 API、需要快速驗證請求和響應的場景。(Traffine I/O)

  • 📘 可選的 API 文檔(ReDoc)
  • 訪問地址:默認情況下,運行 FastAPI 應用后,可以通過 http://127.0.0.1:8000/redoc 訪問。(菜鳥教程)

  • 主要特點

    • 文檔導向:側重于提供詳細的 API 描述,適合生成正式的 API 文檔。
    • 界面簡潔:采用單欄布局,目錄結構清晰,便于查閱。
    • 缺乏交互測試:不支持直接在文檔中測試 API 請求。(菜鳥教程)
  • 適用場景:需要為用戶或第三方開發者提供正式、可讀性強的 API 文檔的場景。

特性Swagger UI(/docs)ReDoc(/redoc)
交互性? 支持直接測試 API 請求? 不支持交互測試
文檔結構📄 卡片式展示,適合快速瀏覽和測試📘 側邊目錄,適合詳細查閱
界面風格🎨 現代化、功能導向🧾 簡潔、文檔導向
適用階段🧪 開發和調試階段📚 正式文檔發布階段(FastAPI)

? 建議

  • 開發階段:使用 Swagger UI(/docs)進行 API 測試和調試。
  • 文檔發布:使用 ReDoc(/redoc)生成正式的 API 文檔供用戶查閱。(菜鳥教程)

FastAPI 通過這兩種文檔界面,滿足了開發者在不同階段的需求,提升了開發效率和文檔質量。

Reqeust

在 FastAPI 中,Request 對象是處理 HTTP 請求的核心組件。它并非 FastAPI 原生實現,而是直接繼承自 Starlette 框架的 Request 類。Starlette 是一個輕量級的 ASGI(Asynchronous Server Gateway Interface)框架,專為構建異步 Web 應用設計。FastAPI 基于 Starlette 構建,提供了數據驗證、依賴注入和自動生成 API 文檔等高級功能。 (知乎專欄, biaodianfu.com)

在 Starlette 中,Request 對象的初始化方式如下:

class Request:def __init__(self, scope, receive):self.scope = scopeself._receive = receiveself._stream_consumed = False

這里的 scope 是一個包含請求信息的字典,符合 ASGI 規范,包含了請求的方法、路徑、頭信息等。receive 是一個異步函數,用于接收請求體的數據。這種設計使得 Request 對象能夠在異步環境中高效地處理 HTTP 請求。

與 Flask 不同,FastAPI(通過 Starlette)并未使用 threading.localcontextvars 來管理請求上下文。相反,它依賴于顯式傳遞的 scopereceive,避免了線程或協程本地存儲的復雜性,提升了性能和可維護性。

路徑參數

FastAPI 支持使用 Python 字符串格式化語法聲明路徑參數(變量):

from fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}")
async def read_item(item_id):return {"item_id": item_id}

這段代碼把路徑參數 item_id 的值傳遞給路徑函數的參數 item_id。

還可以使用 Python 標準類型注解,聲明路徑操作函數中路徑參數的類型(復雜參數校驗應該交給pydantic!):

@app.get("/items/{item_id}")
async def read_item(item_id: int):return {"item_id": item_id}

運行示例并訪問 http://127.0.0.1:8000/items/3注意url位置參數是字符串哦),返回的響應如下:

{"item_id":3}

可以看到,FastAPI 通過類型聲明自動解析請求中的數據,對于不能類型轉換的參數(訪問:http://127.0.0.1:8000/items/foo)則會報錯:

在這里插入圖片描述
順序很重要,有時,路徑操作中的路徑是寫死的。

  1. 比如要使用 /users/me 獲取當前用戶的數據。

  2. 然后還要使用 /users/{user_id},通過用戶 ID 獲取指定用戶的數據。

由于路徑操作是按順序依次運行的,因此,一定要在 /users/{user_id} 之前聲明 /users/me

from fastapi import FastAPIapp = FastAPI()@app.get("/users/me")
async def read_user_me():return {"user_id": "the current user"}@app.get("/users/{user_id}")
async def read_user(user_id: str):return {"user_id": user_id}

否則,/users/{user_id} 將匹配 /users/me,FastAPI 會認為正在接收值為 “me” 的 user_id 參數。

Enum 類用于路徑參數

路徑操作使用 Python 的 Enum 類型接收預設的路徑參數。

導入 Enum 并創建繼承自 str 和 Enum 的子類。通過從 str 繼承,API 文檔就能把值的類型定義為字符串,并且能正確渲染。

然后,創建包含固定值的類屬性,這些固定值是可用的有效值:

from enum import Enumfrom fastapi import FastAPIclass ModelName(str, Enum):alexnet = "alexnet"resnet = "resnet"lenet = "lenet"app = FastAPI()@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):if model_name is ModelName.alexnet:return {"model_name": model_name, "message": "Deep Learning FTW!"}if model_name.value == "lenet":return {"model_name": model_name, "message": "LeCNN all the images"}return {"model_name": model_name, "message": "Have some residuals"}

API 文檔會顯示預定義路徑參數的可用值:

在這里插入圖片描述
可以使用 model_name.value 或 your_enum_member.value 或 ModelName.lenet.value 獲取實際的值(本例中為字符串)!

路徑參數和數值校驗

你可以使用 Path 為路徑參數聲明型校驗和元數據。

from typing import Annotatedfrom fastapi import FastAPI, Path, Queryapp = FastAPI()@app.get("/items/{item_id}")
async def read_items(item_id: Annotated[int, Path(title="The ID of the item to get")],q: Annotated[str | None, Query(alias="item-query")] = None,
):results = {"item_id": item_id}if q:results.update({"q": q})return results

對 FastAPI 來說參數順序無關緊要。它將通過參數的名稱、類型和默認值聲明(Query、Path 等)來檢測參數,而不在乎參數的順序。

因此,你可以將函數聲明為:

from fastapi import FastAPI, Pathapp = FastAPI()@app.get("/items/{item_id}")
async def read_items(q: str, item_id: int = Path(title="The ID of the item to get")):results = {"item_id": item_id}if q:results.update({"q": q})return results

查詢參數

聲明的參數不是路徑參數時,路徑操作函數會把該參數自動解釋為查詢參數。查詢字符串是鍵值對的集合,這些鍵值對位于 URL 的 ? 之后,以 & 分隔。例如,以下 URL 中:

http://127.0.0.1:8000/items/?skip=0&limit=10

查詢參數為:

  • skip:值為 0
  • limit:值為 10

這些值都是 URL 的組成部分,因此,它們的類型本應是字符串。但聲明 Python 類型(上例中為 int)之后,這些值就會轉換為聲明的類型,并進行類型校驗。

from fastapi import FastAPIapp = FastAPI()fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):return fake_items_db[skip : skip + limit]

查詢參數不是路徑的固定內容,它是可選的,還支持默認值。上例用 skip=0 和 limit=10 設定默認值。

訪問 URL:

http://127.0.0.1:8000/items/

與訪問以下地址相同:

http://127.0.0.1:8000/items/?skip=0&limit=10

同理,把默認值設為 None 即可聲明可選的查詢參數,這其實是一種聯合類型聯合類型(Union types),它的意思是變量 q 的類型可以是 str 或者 None,也就是可選的字符串。:

@app.get("/items/{item_id}")
async def read_item(item_id: str, q: str | None = None):if q:return {"item_id": item_id, "q": q}return {"item_id": item_id}

這是 Python 3.10+ 引入的 聯合類型 簡寫形式,相當于:

from typing import Unionq: Union[str, None] = None

python引入類型注解后,我們會經常見到C類型及其擴展類型,標注類型不得不說是一個非常好的開發習慣,盡管Python解釋器并不會校驗類型:

概念示例用途
聯合類型`strNoneUnion[str, None]`變量可以是多個類型中的一個
泛型List[int], Dict[str, Any]參數化類型,如列表中元素是 int
類型別名UserID = int為某個類型起一個新名字,增加可讀性

參數還可以聲明為 bool 類型,FastAPI 會自動轉換參數類型:

from fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}")
async def read_item(item_id: str, q: str | None = None, short: bool = False):item = {"item_id": item_id}if q:item.update({"q": q})if not short:item.update({"description": "This is an amazing item that has a long description"})return item

本例中,訪問:

http://127.0.0.1:8000/items/foo?short=1
或http://127.0.0.1:8000/items/foo?short=True
或http://127.0.0.1:8000/items/foo?short=true
或http://127.0.0.1:8000/items/foo?short=on
或http://127.0.0.1:8000/items/foo?short=yes

或其它任意大小寫形式(大寫、首字母大寫等),函數接收的 short 參數都是布爾值 True。值為 False 時也一樣。

FastAPI 可以識別同時聲明的多個路徑參數和查詢參數,而且聲明查詢參數的順序并不重要。

from fastapi import FastAPIapp = FastAPI()@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(user_id: int, item_id: str, q: str | None = None, short: bool = False
):item = {"item_id": item_id, "owner_id": user_id}if q:item.update({"q": q})if not short:item.update({"description": "This is an amazing item that has a long description"})return item

如果要把查詢參數設置為必選,就不要聲明默認值:

from fastapi import FastAPIapp = FastAPI()@app.get("/items/{item_id}")
async def read_user_item(item_id: str, needy: str):item = {"item_id": item_id, "needy": needy}return item

查詢參數和字符串校驗

FastAPI 允許你為參數聲明額外的信息和校驗,比如我們打算添加約束條件:即使 q 是可選的,但只要提供了該參數,則該參數值不能超過50個字符的長度。

為此,首先從 fastapi 導入 Query:

from typing import Unionfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, max_length=50)):results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}if q:results.update({"q": q})return results

from typing import Annotatedfrom fastapi import FastAPI, Queryapp = FastAPI()@app.get("/items/")
async def read_items(q: Annotated[str | None, Query(max_length=50)] = None):results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}if q:results.update({"q": q})return results

Annotated 是 Python 3.9+ 引入的一個類型提示增強工具,主要作用是給類型添加“元信息”(metadata),使得類型提示不僅表達類型本身,還能攜帶額外說明。

你還可以添加 min_length 參數,正則表達式:

@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, min_length=3, max_length=50, pattern="^fixedquery$"),
):

當你使用 Query 顯式地定義查詢參數時,你還可以聲明它去接收一組值,或換句話來說,接收多個值。

@app.get("/items/")
async def read_items(q: Union[List[str], None] = Query(default=None)):query_items = {"q": q}return query_items

然后請求http://localhost:8000/items/?q=foo&q=bar,你會在路徑操作函數的函數參數 q 中以一個 Python list 的形式接收到查詢參數 q 的多個值。

請求體

使用 Pydantic 模型聲明請求體,能充分利用它的功能和優點。

from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneapp = FastAPI()@app.post("/items/")
async def create_item(item: Item):return item

與聲明查詢參數一樣,包含默認值的模型屬性是可選的,否則就是必選的。默認值為 None 的模型屬性也是可選的。

僅使用 Python 類型聲明,FastAPI 就可以:

  • 以 JSON 形式讀取請求體:(在必要時)把請求體轉換為對應的類型
  • 校驗數據:數據無效時返回錯誤信息,并指出錯誤數據的確切位置和內容
  • 把接收的數據賦值給參數 item,把函數中請求體參數的類型聲明為 Item,還能獲得代碼補全等編輯器支持
  • 為模型生成 JSON Schema,在項目中所需的位置使用。

FastAPI 支持同時聲明路徑參數和請求體,FastAPI 能識別與路徑參數匹配的函數參數,還能識別從請求體中獲取的類型為 Pydantic 模型的函數參數。

from fastapi import FastAPI
from pydantic import BaseModelclass Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneapp = FastAPI()@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):return {"item_id": item_id, **item.dict()}

FastAPI 還支持同時聲明請求體、路徑參數和查詢參數。

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, q: str | None = None):result = {"item_id": item_id, **item.dict()}if q:result.update({"q": q})return result

函數參數按如下規則進行識別:

  • 路徑中聲明了相同參數的參數,是路徑參數
  • 類型是(int、float、str、bool 等)單類型的參數,是查詢參數
  • 類型是 Pydantic 模型的參數,是請求體

多個請求體參數

毫無疑問地,你可以隨意地混合使用 Path、Query 和請求體參數聲明,FastAPI 會知道該如何處理。

from typing import Annotatedfrom fastapi import Body, FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Noneclass User(BaseModel):username: strfull_name: str | None = None@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User, importance: Annotated[int, Body()]
):results = {"item_id": item_id, "item": item, "user": user, "importance": importance}return results

在這種情況下,FastAPI 將期望像這樣的請求體:

{"item": {"name": "Foo","description": "The pretender","price": 42.0,"tax": 3.2},"user": {"username": "dave","full_name": "Dave Grohl"},"importance": 5
}

與使用 Query 和 Path 為查詢參數和路徑參數定義額外數據的方式相同,FastAPI 提供了一個同等的 Body,例如,為了擴展先前的模型,你可能決定除了 item 和 user 之外,還想在同一請求體中具有另一個鍵 importance。

如果你就按原樣聲明它,因為它是一個單一值,FastAPI 將假定它是一個查詢參數。但是你可以使用 Body 指示 FastAPI 將其作為請求體的另一個鍵進行處理。

嵌入單個請求體參數

假設你只有一個來自 Pydantic 模型 Item 的請求體參數 item。

默認情況下,FastAPI 將直接期望這樣的請求體。

但是,如果你希望它期望一個擁有 item 鍵并在值中包含模型內容的 JSON,就像在聲明額外的請求體參數時所做的那樣,則可以使用一個特殊的 Body 參數 embed:

item: Item = Body(embed=True)

比如:

from typing import Annotatedfrom fastapi import Body, FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = None@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):results = {"item_id": item_id, "item": item}return results

在這種情況下,FastAPI 將期望像這樣的請求體:

{"item": {"name": "Foo","description": "The pretender","price": 42.0,"tax": 3.2}
}

而不是:

{"name": "Foo","description": "The pretender","price": 42.0,"tax": 3.2
}

Cookie

定義 Cookie 參數與定義 Query 和 Path 參數一樣。

from typing import Annotatedfrom fastapi import Cookie, FastAPIapp = FastAPI()@app.get("/items/")
async def read_items(ads_id: Annotated[str | None, Cookie()] = None):return {"ads_id": ads_id}

如果您有一組相關的 cookie,您可以創建一個 Pydantic 模型來聲明它們。

from typing import Annotatedfrom fastapi import Cookie, FastAPI
from pydantic import BaseModelapp = FastAPI()class Cookies(BaseModel):session_id: strfatebook_tracker: str | None = Nonegoogall_tracker: str | None = None@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):return cookies

FastAPI 將從請求中接收到的 cookie 中提取出每個字段的數據,并提供您定義的 Pydantic 模型。

注意:你不能直接寫 cookies: Cookies,因為 FastAPI 必須通過 Cookie() 明確告訴它從 cookie 中讀取這些字段,否則它不知道你想從哪里獲取這個 Cookies 模型的字段。

這是 FastAPI 的一個核心設計原則:參數類型 + 額外標記(如 Query, Header, Cookie)共同決定參數來源。

在某些特殊使用情況下(可能并不常見),您可能希望限制您想要接收的 cookie。您可以使用 Pydantic 的模型配置來禁止( forbid )任何額外( extra )字段:

from typing import Annotated, Unionfrom fastapi import Cookie, FastAPI
from pydantic import BaseModelapp = FastAPI()class Cookies(BaseModel):model_config = {"extra": "forbid"}session_id: strfatebook_tracker: Union[str, None] = Nonegoogall_tracker: Union[str, None] = None@app.get("/items/")
async def read_items(cookies: Annotated[Cookies, Cookie()]):return cookies

如果客戶嘗試發送一些額外的 cookie,他們將收到錯誤響應。例如,如果客戶端嘗試發送一個值為 good-list-please 的 santa_tracker cookie,客戶端將收到一個錯誤響應,告知他們 santa_tracker cookie 是不允許的:

{"detail": [{"type": "extra_forbidden","loc": ["cookie", "santa_tracker"],"msg": "Extra inputs are not permitted","input": "good-list-please",}]
}

Header

定義 Header 參數的方式與定義 Query、Path、Cookie 參數相同。

from typing import Annotatedfrom fastapi import FastAPI, Headerapp = FastAPI()@app.get("/items/")
async def read_items(user_agent: Annotated[str | None, Header()] = None):return {"User-Agent": user_agent}

Header 比 Path、Query 和 Cookie 提供了更多功能。大部分標準請求頭用連字符分隔,即減號(-)。

但是 user-agent 這樣的變量在 Python 中是無效的。因此,默認情況下,Header 把參數名中的字符由下劃線(_)改為連字符(-)來提取并存檔請求頭 。

同時,HTTP 的請求頭不區分大小寫,可以使用 Python 標準樣式(即 snake_case)進行聲明。因此,可以像在 Python 代碼中一樣使用 user_agent ,無需把首字母大寫為 User_Agent 等形式。

如需禁用下劃線自動轉換為連字符,可以把 Header 的 convert_underscores 參數設置為 False:

from typing import Annotatedfrom fastapi import FastAPI, Headerapp = FastAPI()@app.get("/items/")
async def read_items(strange_header: Annotated[str | None, Header(convert_underscores=False)] = None,
):return {"strange_header": strange_header}

如果您有一組相關的 header 參數,您可以創建一個 Pydantic 模型來聲明它們。

from typing import Annotatedfrom fastapi import FastAPI, Header
from pydantic import BaseModelapp = FastAPI()class CommonHeaders(BaseModel):host: strsave_data: boolif_modified_since: str | None = Nonetraceparent: str | None = Nonex_tag: list[str] = []@app.get("/items/")
async def read_items(headers: Annotated[CommonHeaders, Header()]):return headers

表單

接收的不是 JSON,而是表單字段時,要使用 Form(需要提前下載pip install python-multipart)。

from fastapi import FastAPI, Formapp = FastAPI()@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):return {"username": username}

創建表單(Form)參數的方式與 Body 和 Query 一樣,例如,OAuth2 規范的 “密碼流” 模式規定要通過表單字段發送 username 和 password。該規范要求字段必須命名為 username 和 password,并通過表單字段發送,不能用 JSON。

使用 Form 可以聲明與 Body (及 Query、Path、Cookie)相同的元數據和驗證。

您可以使用 Pydantic 模型在 FastAPI 中聲明表單字段。您只需聲明一個 Pydantic 模型,其中包含您希望接收的表單字段,然后將參數聲明為 Form :

from typing import Annotatedfrom fastapi import FastAPI, Form
from pydantic import BaseModelapp = FastAPI()class FormData(BaseModel):username: strpassword: str@app.post("/login/")
async def login(data: Annotated[FormData, Form()]):return data

文件

File 用于定義客戶端的上傳文件,因為上傳文件以「表單數據」形式發送,所以接收上傳文件,要預先安裝 python-multipart。

from fastapi import FastAPI, File, UploadFileapp = FastAPI()@app.post("/files/")
async def create_file(file: bytes = File()):return {"file_size": len(file)}@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):return {"filename": file.filename}

如果把路徑操作函數參數的類型聲明為 bytes,FastAPI 將以 bytes 形式讀取和接收文件內容。這種方式把文件的所有內容都存儲在內存里,適用于小型文件。不過,很多情況下,UploadFile 更好用。

UploadFile 與 bytes 相比有更多優勢:

  • 使用 spooled 文件:存儲在內存的文件超出最大上限時,FastAPI 會把文件存入磁盤;
  • 這種方式更適于處理圖像、視頻、二進制文件等大型文件,好處是不會占用所有內存;
  • 可獲取上傳文件的元數據;
  • 自帶 file-like async 接口;
  • 暴露的 Python SpooledTemporaryFile 對象,可直接傳遞給其他預期「file-like」對象的庫。

UploadFile 對象包含以下常用屬性:

屬性名類型描述
filenamestr上傳文件的文件名,例如 "myimage.jpg"
content_typestr文件的 MIME 類型,例如 "image/jpeg"
fileSpooledTemporaryFile(類文件對象)可像普通 Python 文件一樣使用,支持傳遞給支持 file-like 的其他庫

UploadFile 提供了一系列基于異步的文件操作方法,這些方法底層由 SpooledTemporaryFile 提供支持:

方法名參數描述
await write(data)strbytes將數據寫入文件
await read(size)int(可選)讀取指定大小(字節數)的內容,默認讀取全部
await seek(offset)int移動文件指針至指定位置,例如 await file.seek(0) 移動到文件開頭
await close()關閉文件對象

異步讀取示例(在 async def 中)

@app.post("/upload/")
async def upload_file(myfile: UploadFile):contents = await myfile.read()  # 異步讀取上傳內容return {"filename": myfile.filename, "content": contents.decode()}

同步讀取示例(在 def 中)

@app.post("/upload/")
def upload_file(myfile: UploadFile):contents = myfile.file.read()  # 同步讀取return {"filename": myfile.filename, "content": contents.decode()}

💡 注意事項

  • UploadFile 是 FastAPI 推薦用于處理大文件上傳的方式,優于 bytes,因為它不會將整個文件內容加載到內存中。
  • 方法為異步(async),必須在 async def 函數中使用 await
  • 如果你用的是同步路徑操作函數(def),可以直接訪問 .file,像普通文件對象一樣操作。
  • 使用 async 方法時,FastAPI 在線程池中執行文件方法,并 await 操作完成。

您可以通過使用標準類型注解并將 None 作為默認值的方式將一個文件參數設為可選:

from fastapi import FastAPI, File, UploadFileapp = FastAPI()@app.post("/files/")
async def create_file(file: bytes | None = File(default=None)):if not file:return {"message": "No file sent"}else:return {"file_size": len(file)}@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile | None = None):if not file:return {"message": "No upload file sent"}else:return {"filename": file.filename}

您也可以將 File() 與 UploadFile 一起使用,例如,設置額外的元數據:

from fastapi import FastAPI, File, UploadFileapp = FastAPI()@app.post("/files/")
async def create_file(file: bytes = File(description="A file read as bytes")):return {"file_size": len(file)}@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(description="A file read as UploadFile"),
):return {"filename": file.filename}

FastAPI 支持同時上傳多個文件。可用同一個「表單字段」發送含多個文件的「表單數據」。上傳多個文件時,要聲明含 bytes 或 UploadFile 的列表(List):

@app.post("/files/")
async def create_files(files: list[bytes] = File()):return {"file_sizes": [len(file) for file in files]}@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile]):return {"filenames": [file.filename for file in files]}@app.post("/files/")
async def create_files(files: list[bytes] = File(description="Multiple files as bytes"),
):return {"file_sizes": [len(file) for file in files]}@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile] = File(description="Multiple files as UploadFile"),
):return {"filenames": [file.filename for file in files]}

直接使用請求

至此,我們已經使用多種類型聲明了請求的各種組件。

并從以下對象中提取數據:

  • 路徑參數
  • 請求頭
  • Cookies

FastAPI 使用這種方式驗證數據、轉換數據,并自動生成 API 文檔。但有時,我們也需要直接訪問 Request 對象。

FastAPI 的底層是 Starlette,FastAPI 只不過是在 Starlette 頂層提供了一些工具,所以能直接使用 Starlette 的 Request 對象。

假設要在路徑操作函數中獲取客戶端 IP 地址和主機,此時,需要直接訪問請求。

from fastapi import FastAPI, Requestapp = FastAPI()@app.get("/items/{item_id}")
def read_root(item_id: str, request: Request):client_host = request.client.hostreturn {"client_host": client_host, "item_id": item_id}

把路徑操作函數的參數類型聲明為 Request,FastAPI 就能把 Request 傳遞到參數里。

Response

在 FastAPI 中,響應(Response) 是與請求(Request)相對的核心部分,用于將數據、安全信息、狀態碼和內容格式傳遞給客戶端。

FastAPI 默認會根據返回值自動生成響應內容,例如 dict 會被自動轉換為 JSON 并附帶合適的 Content-Type。但當你需要自定義響應格式、狀態碼、頭部或流式響應時,可以使用 FastAPI 提供的 Response 類及其衍生類,如:

  • JSONResponse:返回 JSON 數據(默認類型);
  • HTMLResponse:返回 HTML 頁面內容;
  • PlainTextResponse:返回純文本;
  • StreamingResponse:流式傳輸大文件或數據;
  • RedirectResponse:用于重定向;
  • FileResponse:用于文件下載;
  • ORJSONResponse / UJSONResponse:使用更快的 JSON 序列化庫以提升性能。

此外,FastAPI 允許你通過:

  • 設置響應模型(response_model):對輸出內容結構進行約束與驗證;
  • 使用 Response 參數:動態修改響應頭、狀態碼、cookie 等;
  • 自定義狀態碼與媒體類型:精細控制返回行為。

Response Model

你可以在任意的路徑操作中使用 response_model 參數來聲明用于響應的模型:

  • @app.get()
  • @app.post()
  • @app.put()
  • @app.delete()
  • 等等。

并且,response_model是「裝飾器」方法(get,post 等)的一個參數。不像之前的所有參數和請求體,它不屬于路徑操作函數。

它接收的類型與你將為 Pydantic 模型屬性所聲明的類型相同,因此它可以是一個 Pydantic 模型,但也可以是一個由 Pydantic 模型組成的 list,例如 List[Item]。

FastAPI 將使用此 response_model 來:

  • 將輸出數據轉換為其聲明的類型。
  • 校驗數據。
  • 在 OpenAPI 的路徑操作中為響應添加一個 JSON Schema。
  • 并在自動生成文檔系統中使用。
  • 將輸出數據限制在該模型定義內
from typing import Anyfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = Nonetags: list[str] = []@app.post("/items/", response_model=Item)
async def create_item(item: Item) -> Any:return item@app.get("/items/", response_model=list[Item])
async def read_items() -> Any:return [{"name": "Portal Gun", "price": 42.0},{"name": "Plumbus", "price": 32.0},]

你的響應模型可以具有默認值,例如:

from typing import List, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: float = 10.5tags: List[str] = []items = {"foo": {"name": "Foo", "price": 50.2},"bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},"baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
}@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)
async def read_item(item_id: str):return items[item_id]

但如果它們并沒有存儲實際的值,你可能想從結果中忽略它們的默認值。你可以設置路徑操作裝飾器的 response_model_exclude_unset=True 參數:

@app.get("/items/{item_id}", response_model=Item, response_model_exclude_unset=True)

然后響應中將不會包含那些默認值,而是僅有實際設置的值。

你還可以使用路徑操作裝飾器的 response_model_include 和 response_model_exclude 參數。它們接收一個由屬性名稱 str 組成的 set 來包含(忽略其他的)或者排除(包含其他的)這些屬性。(可以但不建議)

多個關聯模型

下面的代碼展示了不同模型處理密碼字段的方式,及使用位置的大致思路:

from fastapi import FastAPI
from pydantic import BaseModel, EmailStrapp = FastAPI()class UserIn(BaseModel):username: strpassword: stremail: EmailStrfull_name: str | None = Noneclass UserOut(BaseModel):username: stremail: EmailStrfull_name: str | None = Noneclass UserInDB(BaseModel):username: strhashed_password: stremail: EmailStrfull_name: str | None = Nonedef fake_password_hasher(raw_password: str):return "supersecret" + raw_passworddef fake_save_user(user_in: UserIn):hashed_password = fake_password_hasher(user_in.password)user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)print("User saved! ..not really")return user_in_db@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):user_saved = fake_save_user(user_in)return user_saved

Pydantic 模型支持 .dict() 方法,能返回包含模型數據的字典。

FastAPI 的核心思想就是減少代碼重復。代碼重復會導致 bug、安全問題、代碼失步等問題(更新了某個位置的代碼,但沒有同步更新其它位置的代碼)。

上面的這些模型共享了大量數據,擁有重復的屬性名和類型。

聲明 UserBase 模型作為其它模型的基類。然后,用該類衍生出繼承其屬性(類型聲明、驗證等)的子類。所有數據轉換、校驗、文檔等功能仍將正常運行。這樣,就可以僅聲明模型之間的差異部分(具有明文的 password、具有 hashed_password 以及不包括密碼)。

通過這種方式,可以只聲明模型之間的區別(分別包含明文密碼、哈希密碼,以及無密碼的模型)。

from fastapi import FastAPI
from pydantic import BaseModel, EmailStrapp = FastAPI()class UserBase(BaseModel):username: stremail: EmailStrfull_name: str | None = Noneclass UserIn(UserBase):password: strclass UserOut(UserBase):passclass UserInDB(UserBase):hashed_password: strdef fake_password_hasher(raw_password: str):return "supersecret" + raw_passworddef fake_save_user(user_in: UserIn):hashed_password = fake_password_hasher(user_in.password)user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password)print("User saved! ..not really")return user_in_db@app.post("/user/", response_model=UserOut)
async def create_user(user_in: UserIn):user_saved = fake_save_user(user_in)return user_saved

任意的 dict 都能用于聲明響應,只要聲明鍵和值的類型,無需使用 Pydantic 模型。事先不知道可用的字段 / 屬性名時(Pydantic 模型必須知道字段是什么),這種方式特別有用。

from fastapi import FastAPIapp = FastAPI()@app.get("/keyword-weights/", response_model=dict[str, float])
async def read_keyword_weights():return {"foo": 2.3, "bar": 3.4}

響應狀態碼

與指定響應模型的方式相同,在以下任意路徑操作中,可以使用 status_code 參數聲明用于響應的 HTTP 狀態碼:

  • @app.get()
  • @app.post()
  • @app.put()
  • @app.delete()
  • 等……

status_code 是(get、post 等)裝飾器方法中的參數。與之前的參數和請求體不同,不是路徑操作函數的參數。

from fastapi import FastAPIapp = FastAPI()@app.post("/items/", status_code=201)
async def create_item(name: str):return {"name": name}

status_code 還能接收 IntEnum 類型,比如 Python 的 http.HTTPStatus。

from fastapi import FastAPI, statusapp = FastAPI()@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):return {"name": name}

在 HTTP 協議中,發送 3 位數的數字狀態碼是響應的一部分。這些狀態碼都具有便于識別的關聯名稱,但是重要的還是數字。

簡言之:

  • 100 及以上的狀態碼用于返回信息。這類狀態碼很少直接使用。具有這些狀態碼的響應不能包含響應體。

  • 200 及以上的狀態碼用于表示成功。這些狀態碼是最常用的。

    • 200 是默認狀態代碼,表示一切正常。
    • 201 表示已創建,通常在數據庫中創建新記錄后使用。
    • 204 是一種特殊的例子,表示無內容。該響應在沒有為客戶端返回內容時使用,因此,該響應不能包含響應體。
  • 300 及以上的狀態碼用于重定向。具有這些狀態碼的響應不一定包含響應體,但 304 未修改 是個例外,該響應不得包含響應體。

  • 400 及以上的狀態碼用于表示客戶端錯誤。這些可能是第二常用的類型。

    • 404 用于未找到響應。
    • 對于來自客戶端的一般錯誤,可以只使用 400
  • 500 及以上的狀態碼用于表示服務器端錯誤。幾乎永遠不會直接使用這些狀態碼。應用代碼或服務器出現問題時,會自動返回這些狀態代碼。

處理錯誤

某些情況下,需要向客戶端返回錯誤提示。這里所謂的客戶端包括前端瀏覽器、其他應用程序、物聯網設備等。

需要向客戶端返回錯誤提示的場景主要如下:

  • 客戶端沒有執行操作的權限
  • 客戶端沒有訪問資源的權限
  • 客戶端要訪問的項目不存在
  • 等等 …

向客戶端返回 HTTP 錯誤響應,可以使用 HTTPException。HTTPException 是額外包含了和 API 有關數據的常規 Python 異常。

因為是 Python 異常,所以不能 return,只能 raise。

如在調用函數里的工具函數時,觸發了 HTTPException,FastAPI 就不再繼續執行路徑操作函數中的后續代碼,而是立即終止請求,并把 HTTPException 的 HTTP 錯誤發送至客戶端。

from fastapi import FastAPI, HTTPExceptionapp = FastAPI()items = {"foo": "The Foo Wrestlers"}@app.get("/items/{item_id}")
async def read_item(item_id: str):if item_id not in items:raise HTTPException(status_code=404, detail="Item not found")return {"item": items[item_id]}

如果客戶端請求 http://example.com/items/bar(item_id 「bar」 不存在時),則會接收到 HTTP 狀態碼 - 404(「未找到」錯誤)及如下 JSON 響應結果:

{"detail": "Item not found"
}

觸發 HTTPException 時,可以用參數 detail 傳遞任何能轉換為 JSON 的值,不僅限于 str。還支持傳遞 dict、list 等數據結構。FastAPI 能自動處理這些數據,并將之轉換為 JSON。

對于某些高級應用場景,還需要添加自定義響應頭:

from fastapi import FastAPI, HTTPExceptionapp = FastAPI()items = {"foo": "The Foo Wrestlers"}@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):if item_id not in items:raise HTTPException(status_code=404,detail="Item not found",headers={"X-Error": "There goes my error"},)return {"item": items[item_id]}

自定義異常處理

添加自定義處理器,要使用 Starlette 的異常工具。

假設要觸發的自定義異常叫作 UnicornException。且需要 FastAPI 實現全局處理該異常。此時,可以用 @app.exception_handler() 添加自定義異常控制器:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponseclass UnicornException(Exception):def __init__(self, name: str):self.name = nameapp = FastAPI()@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):return JSONResponse(status_code=418,content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},)@app.get("/unicorns/{name}")
async def read_unicorn(name: str):if name == "yolo":raise UnicornException(name=name)return {"unicorn_name": name}

請求 /unicorns/yolo 時,路徑操作會觸發 UnicornException。但該異常將會被 unicorn_exception_handler 處理。接收到的錯誤信息清晰明了,HTTP 狀態碼為 418,JSON 內容如下:

{"message": "Oops! yolo did something. There goes a rainbow..."}

FastAPI 自帶了一些默認異常處理器。觸發 HTTPException 或請求無效數據時,這些處理器返回默認的 JSON 響應結果。不過,也可以使用自定義處理器覆蓋默認異常處理器。

請求中包含無效數據時,FastAPI 內部會觸發 RequestValidationError。該異常也內置了默認異常處理器。覆蓋默認異常處理器時需要導入 RequestValidationError,并用 @app.excption_handler(RequestValidationError) 裝飾異常處理器。這樣,異常處理器就可以接收 Request 與異常。

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPExceptionapp = FastAPI()@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):return PlainTextResponse(str(exc.detail), status_code=exc.status_code)@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):return PlainTextResponse(str(exc), status_code=400)@app.get("/items/{item_id}")
async def read_item(item_id: int):if item_id == 3:raise HTTPException(status_code=418, detail="Nope! I don't like 3.")return {"item_id": item_id}

RequestValidationError 是 Pydantic 的 ValidationError 的子類。

FastAPI 調用的就是 RequestValidationError 類,因此,如果在 response_model 中使用 Pydantic 模型,且數據有錯誤時,在日志中就會看到這個錯誤。

但客戶端或用戶看不到這個錯誤。反之,客戶端接收到的是 HTTP 狀態碼為 500 的「內部服務器錯誤」。這是因為在響應或代碼(不是在客戶端的請求里)中出現的 Pydantic ValidationError 是代碼的 bug。

同理,也可以覆蓋 HTTPException 處理器。例如,只為錯誤返回純文本響應,而不是返回 JSON 格式的內容:

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPExceptionapp = FastAPI()@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):return PlainTextResponse(str(exc.detail), status_code=exc.status_code)@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):return PlainTextResponse(str(exc), status_code=400)@app.get("/items/{item_id}")
async def read_item(item_id: int):if item_id == 3:raise HTTPException(status_code=418, detail="Nope! I don't like 3.")return {"item_id": item_id}

RequestValidationError 包含其接收到的無效數據請求的 body 。開發時,可以用這個請求體生成日志、調試錯誤,并返回給用戶。

from fastapi import FastAPI, Request, status
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModelapp = FastAPI()@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):return JSONResponse(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}),)class Item(BaseModel):title: strsize: int@app.post("/items/")
async def create_item(item: Item):return item

FastAPI 也提供了自有的 HTTPException。FastAPI 的 HTTPException 繼承自 Starlette 的 HTTPException 錯誤類。

它們之間的唯一區別是,FastAPI 的 HTTPException 可以在響應中添加響應頭。OAuth 2.0 等安全工具需要在內部調用這些響應頭。

因此你可以繼續像平常一樣在代碼中觸發 FastAPI 的 HTTPException 。但注冊異常處理器時,應該注冊到來自 Starlette 的 HTTPException。

這樣做是為了,當 Starlette 的內部代碼、擴展或插件觸發 Starlette HTTPException 時,處理程序能夠捕獲、并處理此異常。

FastAPI 還支持先對異常進行某些處理,然后再使用 FastAPI 中處理該異常的默認異常處理器。

從 fastapi.exception_handlers 中導入要復用的默認異常處理器:

from fastapi import FastAPI, HTTPException
from fastapi.exception_handlers import (http_exception_handler,request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPExceptionapp = FastAPI()@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request, exc):print(f"OMG! An HTTP error!: {repr(exc)}")return await http_exception_handler(request, exc)@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):print(f"OMG! The client sent invalid data!: {exc}")return await request_validation_exception_handler(request, exc)@app.get("/items/{item_id}")
async def read_item(item_id: int):if item_id == 3:raise HTTPException(status_code=418, detail="Nope! I don't like 3.")return {"item_id": item_id}

直接返回響應

當你創建一個 FastAPI 路徑操作 時,你可以正常返回以下任意一種數據:dict,list,Pydantic 模型,數據庫模型等等。

FastAPI 默認會使用 jsonable_encoder 將這些類型的返回值轉換成 JSON 格式。然后,FastAPI 會在后臺將這些兼容 JSON 的數據(比如字典)放到一個 JSONResponse 中,該 JSONResponse 會用來發送響應給客戶端。但是你可以在你的 路徑操作 中直接返回一個 JSONResponse。

直接返回響應可能會有用處,比如返回自定義的響應頭和 cookies。

**事實上,你可以返回任意 Response 或者任意 Response 的子類。**當你返回一個 Response 時,FastAPI 會直接傳遞它。

FastAPI 不會用 Pydantic 模型做任何數據轉換,不會將響應內容轉換成任何類型,等等。由于 FastAPI 并未對你返回的 Response 做任何改變,你必須確保你已經準備好響應內容。

例如,如果不首先將 Pydantic 模型轉換為 dict,并將所有數據類型(如 datetime、UUID 等)轉換為兼容 JSON 的類型,則不能將其放入JSONResponse中。

from datetime import datetime
from typing import Unionfrom fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse
from pydantic import BaseModelclass Item(BaseModel):title: strtimestamp: datetimedescription: Union[str, None] = Noneapp = FastAPI()@app.put("/items/{id}")
def update_item(id: str, item: Item):json_compatible_item_data = jsonable_encoder(item)return JSONResponse(content=json_compatible_item_data)

假設你想要返回一個 XML 響應。你可以把你的 XML 內容放到一個字符串中,放到一個 Response 中,然后返回。

from fastapi import FastAPI, Responseapp = FastAPI()@app.get("/legacy/")
def get_legacy_data():data = """<?xml version="1.0"?><shampoo><Header>Apply shampoo here.</Header><Body>You'll have to use soap here.</Body></shampoo>"""return Response(content=data, media_type="application/xml")

FastAPI 默認使用 JSONResponse 返回一個響應,將你的視圖函數中的返回內容放到該 JSONResponse 中。FastAPI 會自動使用默認的狀態碼或者使用你在 路徑操作 中設置的狀態碼。如果你想要返回主要狀態碼之外的狀態碼:

return JSONResponse(status_code=status.HTTP_201_CREATED, content=item)

HTML,流,文件和其他

例如,如果你需要壓榨性能,你可以安裝并使用 orjson 并將響應設置為 ORJSONResponse。

from fastapi import FastAPI
from fastapi.responses import ORJSONResponseapp = FastAPI()@app.get("/items/", response_class=ORJSONResponse)
async def read_items():return ORJSONResponse([{"item_id": "Foo"}])

注意:ORJSONResponse 目前只在 FastAPI 中可用,而在 Starlette 中不可用。

如果你需要使用 HTMLResponse 來從 FastAPI 中直接返回一個 HTML 響應。導入 HTMLResponse,將 HTMLResponse 作為你的 路徑操作 的 response_class 參數傳入。

from fastapi import FastAPI
from fastapi.responses import HTMLResponseapp = FastAPI()@app.get("/items/", response_class=HTMLResponse)
async def read_items():return """<html><head><title>Some HTML in here</title></head><body><h1>Look ma! HTML!</h1></body></html>"""

或者也可以直接返回一個HTMLResponse:

from fastapi import FastAPI
from fastapi.responses import HTMLResponseapp = FastAPI()@app.get("/items/")
async def read_items():html_content = """<html><head><title>Some HTML in here</title></head><body><h1>Look ma! HTML!</h1></body></html>"""return HTMLResponse(content=html_content, status_code=200)

返回 HTTP 重定向。默認情況下使用 307 狀態代碼(臨時重定向)。

from fastapi import FastAPI
from fastapi.responses import RedirectResponseapp = FastAPI()@app.get("/typer")
async def redirect_typer():return RedirectResponse("https://typer.tiangolo.com")

要傳輸流式傳輸響應主體,需要采用異步生成器或普通生成器/迭代器:

from fastapi import FastAPI
from fastapi.responses import StreamingResponseapp = FastAPI()async def fake_video_streamer():for i in range(10):yield b"some fake video bytes"@app.get("/")
async def main():return StreamingResponse(fake_video_streamer())

如果您有類似文件的對象(例如,由 open() 返回的對象),則可以在 StreamingResponse 中將其返回。包括許多與云存儲,視頻處理等交互的庫。

from fastapi import FastAPI
from fastapi.responses import StreamingResponsesome_file_path = "large-video-file.mp4"
app = FastAPI()@app.get("/")
def main():def iterfile():  # (1)with open(some_file_path, mode="rb") as file_like:  # (2)yield from file_like  # (3)return StreamingResponse(iterfile(), media_type="video/mp4")

如果需要異步傳輸文件作為響應則可以使用FileResponse,

  • path - 要流式傳輸的文件的文件路徑。
  • headers - 任何自定義響應頭,傳入字典類型。
  • media_type - 給出媒體類型的字符串。如果未設置,則文件名或路徑將用于推斷媒體類型。
  • filename - 如果給出,它將包含在響應的 Content-Disposition 中。
  • 文件響應將包含適當的 Content-Length,Last-Modified 和 ETag 的響應頭。
from fastapi import FastAPI
from fastapi.responses import FileResponsesome_file_path = "large-video-file.mp4"
app = FastAPI()@app.get("/")
async def main():return FileResponse(some_file_path)

StreamingResponseFileResponse 都是 FastAPI 中用于返回文件或流式數據的響應類,但它們的使用場景和底層行為有所不同。

  • StreamingResponse
    用于返回一個流式響應,其內容可以是任何可迭代(包括生成器)或 async 可迭代對象(如異步生成器),適合用于返回動態生成的大文件或實時數據傳輸,節省內存開銷。例如,當你要逐塊讀取大文件、實時編碼視頻流、返回日志輸出等場景,可以使用 StreamingResponse

  • FileResponse
    專為返回靜態文件設計。它會自動處理如文件大小、媒體類型、內容編碼、瀏覽器緩存頭等細節。適用于你已經有一個本地文件,并希望將其作為下載或直接返回給客戶端的場景。底層使用 aiofiles 異步讀取文件內容,性能優良且使用簡單。

簡而言之:

  • 如果你返回的是一個本地文件路徑 —— 用 FileResponse
  • 如果你返回的是一個生成器或動態數據流 —— 用 StreamingResponse

Cookies

你可以在 路徑函數 中定義一個類型為 Response的參數,這樣你就可以在這個臨時響應對象中設置cookie了。

from fastapi import FastAPI, Responseapp = FastAPI()@app.post("/cookie-and-object/")
def create_cookie(response: Response):response.set_cookie(key="fakesession", value="fake-cookie-session-value")return {"message": "Come to the dark side, we have cookies"}

如果你定義了 response_model,程序會自動根據response_model來過濾和轉換你響應的對象。

響應頭

你可以在你的路徑操作函數中聲明一個Response類型的參數(就像你可以為cookies做的那樣)。然后你可以在這個臨時響應對象中設置頭部。

from fastapi import FastAPI, Responseapp = FastAPI()@app.get("/headers-and-object/")
def get_headers(response: Response):response.headers["X-Cat-Dog"] = "alone in the world"return {"message": "Hello World"}

然后你可以像平常一樣返回任何你需要的對象(例如一個dict或者一個數據庫模型)。如果你聲明了一個response_model,它仍然會被用來過濾和轉換你返回的對象。

你也可以在直接返回Response時添加頭部。

from fastapi import FastAPI
from fastapi.responses import JSONResponseapp = FastAPI()@app.get("/headers/")
def get_headers():content = {"message": "Hello World"}headers = {"X-Cat-Dog": "alone in the world", "Content-Language": "en-US"}return JSONResponse(content=content, headers=headers)

路徑操作

在 FastAPI 中,所謂“路徑操作”(Path Operation),指的是將某個 URL 路徑與一個 HTTP 方法(如 GET、POST、PUT 等)綁定,并關聯到一個 Python 函數上的過程。

換句話說,一個“路徑操作”就是你定義的一個 API 端點。它由兩個核心組成部分構成:

  • 路徑(Path):如 /items//users/{user_id} 等 URL 路徑;
  • 操作(Operation):如 GET、POST 等 HTTP 方法。

FastAPI 通過裝飾器(如 @app.get()@app.post() 等)來聲明路徑操作。例如:

@app.get("/items/{item_id}")
async def read_item(item_id: int):return {"item_id": item_id}

上面的代碼就定義了一個路徑為 /items/{item_id},方法為 GET 的路徑操作函數 read_item()

FastAPI 會根據路徑、方法、函數簽名等信息,自動為你生成路由匹配、參數解析、類型校驗、自動文檔等功能。

因此,“路徑操作”是 FastAPI 的核心概念之一,它代表了你 API 的具體業務入口。每一個路徑操作函數,都是一個 Web 接口。

路徑操作配置

路徑操作裝飾器支持多種配置參數。

注意:以下參數應直接傳遞給路徑操作裝飾器,不能傳遞給路徑操作函數。

status_code 用于定義路徑操作響應中的 HTTP 狀態碼。可以直接傳遞 int 代碼, 比如 404。如果記不住數字碼的涵義,也可以用 status 的快捷常量:

from typing import Set, Unionfrom fastapi import FastAPI, status
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):return item

tags 參數的值是由 str 組成的 list (一般只有一個 str ),tags 用于為路徑操作添加標簽:

from typing import Set, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/", response_model=Item, tags=["items"])
async def create_item(item: Item):return item@app.get("/items/", tags=["items"])
async def read_items():return [{"name": "Foo", "price": 42}]@app.get("/users/", tags=["users"])
async def read_users():return [{"username": "johndoe"}]

路徑裝飾器還支持 summary 和 description 這兩個參數:

from typing import Set, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/",response_model=Item,summary="Create an item",description="Create an item with all the information, name, description, price, tax and a set of unique tags",
)
async def create_item(item: Item):return item

描述內容比較長且占用多行時,可以在函數的 docstring 中聲明路徑操作的描述,FastAPI 支持從文檔字符串中讀取描述內容。

from typing import Set, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/", response_model=Item, summary="Create an item")
async def create_item(item: Item):"""Create an item with all the information:- **name**: each item must have a name- **description**: a long description- **price**: required- **tax**: if the item doesn't have tax, you can omit this- **tags**: a set of unique tag strings for this item"""return item

response_description 參數用于定義響應的描述說明:

from typing import Set, Unionfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Union[str, None] = Noneprice: floattax: Union[float, None] = Nonetags: Set[str] = set()@app.post("/items/",response_model=Item,summary="Create an item",response_description="The created item",
)
async def create_item(item: Item):"""Create an item with all the information:- **name**: each item must have a name- **description**: a long description- **price**: required- **tax**: if the item doesn't have tax, you can omit this- **tags**: a set of unique tag strings for this item"""return item

注意,response_description 只用于描述響應,description 一般則用于描述路徑操作。

deprecated 參數可以把路徑操作標記為棄用,無需直接刪除:

from fastapi import FastAPIapp = FastAPI()@app.get("/items/", tags=["items"])
async def read_items():return [{"name": "Foo", "price": 42}]@app.get("/users/", tags=["users"])
async def read_users():return [{"username": "johndoe"}]@app.get("/elements/", tags=["items"], deprecated=True)
async def read_elements():return [{"item_id": "Foo"}]

依賴項

FastAPI 提供了簡單易用,但功能強大的依賴注入系統。

編程中的「依賴注入」是聲明代碼(本文中為路徑操作函數 )運行所需的,或要使用的「依賴」的一種方式。

然后,由系統(本文中為 FastAPI)負責執行任意需要的邏輯,為代碼提供這些依賴(「注入」依賴項)。

依賴注入常用于以下場景:

  • 共享業務邏輯(復用相同的代碼邏輯)
  • 共享數據庫連接
  • 實現安全、驗證、角色權限
  • 等……

上述場景均可以使用依賴注入,將代碼重復最小化。

依賴項就是一個函數,且可以使用與路徑操作函數相同的參數,與在路徑操作函數參數中使用 Body、Query 的方式相同,聲明依賴項需要使用 Depends 和一個新的參數:

from typing import Unionfrom fastapi import Depends, FastAPIapp = 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

依賴項函數的形式和結構與路徑操作函數一樣。因此,可以把依賴項當作沒有「裝飾器」(即,沒有 @app.get(“/some-path”) )的路徑操作函數。

雖然在路徑操作函數的參數中使用 Depends 的方式與 Body、Query 相同,但 Depends 的工作方式略有不同。
這里只能傳給 Depends 一個參數。且該參數必須是可調用對象,比如函數。該函數接收的參數和路徑操作函數的參數一樣。

接收到新的請求時,FastAPI 執行如下操作:

  • 用正確的參數調用依賴項函數(「可依賴項」)
  • 獲取函數返回的結果
  • 把函數返回的結果賦值給路徑操作函數的參數

在這里插入圖片描述
這樣,只編寫一次代碼,FastAPI 就可以為多個路徑操作共享這段代碼 。

比如,下面有 4 個 API 路徑操作(端點):

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

開發人員可以使用依賴項及其子依賴項為這些路徑操作添加不同的權限:

在這里插入圖片描述

類作為依賴項

Python 中的 “可調用對象” 是指任何 Python 可以像函數一樣 “調用” 的對象。在前面的例子中, 我們從依賴項 (“可依賴對象”) 中返回了一個 dict,我們知道編輯器不能為 dict 提供很多支持(比如補全),因為編輯器不知道 dict 的鍵和值類型。

因此,在 FastAPI 中,你可以使用一個 Python 類作為一個依賴項。

from fastapi import Depends, FastAPIapp = 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@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 response

FastAPI 調用 CommonQueryParams 類。這將創建該類的一個 “實例”,該實例將作為參數 commons 被傳遞給你的函數。

子依賴項

FastAPI 支持創建含子依賴項的依賴項。并且,可以按需聲明任意深度的子依賴項嵌套層級。FastAPI 負責處理解析不同深度的子依賴項。

from typing import Unionfrom fastapi import Cookie, Depends, FastAPIapp = FastAPI()def query_extractor(q: Union[str, None] = None):return qdef query_or_cookie_extractor(q: str = Depends(query_extractor),last_query: Union[str, None] = Cookie(default=None),
):if not q:return last_queryreturn q@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):return {"q_or_cookie": query_or_default}

在這里插入圖片描述
如果在同一個路徑操作 多次聲明了同一個依賴項,例如,多個依賴項共用一個子依賴項,FastAPI 在處理同一請求時,只調用一次該子依賴項。

FastAPI 不會為同一個請求多次調用同一個依賴項,而是把依賴項的返回值進行「緩存」,并把它傳遞給同一請求中所有需要使用該返回值的「依賴項」。

在高級使用場景中,如果不想使用「緩存」值,而是為需要在同一請求的每一步操作(多次)中都實際調用依賴項,可以把 Depends 的參數 use_cache 的值設置為 False :

async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)):return {"fresh_value": fresh_value}

路徑操作裝飾器依賴項

有時,我們并不需要在路徑操作函數中使用依賴項的返回值,但仍要執行或解析該依賴項。

對于這種情況,不必在聲明路徑操作函數的參數時使用 Depends,而是可以在路徑操作裝飾器中添加一個由 dependencies 組成的 list。

from fastapi import Depends, FastAPI, Header, HTTPExceptionapp = FastAPI()async def verify_token(x_token: str = Header()):if x_token != "fake-super-secret-token":raise HTTPException(status_code=400, detail="X-Token header invalid")async def verify_key(x_key: str = Header()):if x_key != "fake-super-secret-key":raise HTTPException(status_code=400, detail="X-Key header invalid")return x_key@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():return [{"item": "Foo"}, {"item": "Bar"}]

路徑操作裝飾器依賴項(以下簡稱為“路徑裝飾器依賴項”)的執行或解析方式和普通依賴項一樣,但就算這些依賴項會返回值,它們的值也不會傳遞給路徑操作函數。

全局依賴項

有時,我們要為整個應用添加依賴項。

from fastapi import Depends, FastAPI, Header, HTTPExceptionasync def verify_token(x_token: str = Header()):if x_token != "fake-super-secret-token":raise HTTPException(status_code=400, detail="X-Token header invalid")async def verify_key(x_key: str = Header()):if x_key != "fake-super-secret-key":raise HTTPException(status_code=400, detail="X-Key header invalid")return x_keyapp = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])@app.get("/items/")
async def read_items():return [{"item": "Portal Gun"}, {"item": "Plumbus"}]@app.get("/users/")
async def read_users():return [{"username": "Rick"}, {"username": "Morty"}]

使用yield的依賴項

FastAPI支持在完成后執行一些額外步驟的依賴項,為此,你需要使用 yield 而不是 return,然后再編寫這些額外的步驟(代碼)。

例如,你可以使用這種方式創建一個數據庫會話,并在完成后關閉它。

async def get_db():db = DBSession()try:yield dbfinally:db.close()

你可以聲明任意數量和層級的樹狀依賴,而且它們中的任何一個或所有的都可以使用 yield。

FastAPI 會確保每個帶有 yield 的依賴中的"退出代碼"按正確順序運行。

例如,dependency_c 可以依賴于 dependency_b,而 dependency_b 則依賴于 dependency_a。

from typing import Annotatedfrom fastapi import Dependsasync def dependency_a():dep_a = generate_dep_a()try:yield dep_afinally:dep_a.close()async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]):dep_b = generate_dep_b()try:yield dep_bfinally:dep_b.close(dep_a)async def dependency_c(dep_b: Annotated[DepB, Depends(dependency_b)]):dep_c = generate_dep_c()try:yield dep_cfinally:dep_c.close(dep_b)

同樣,你可以混合使用帶有 yield 或 return 的依賴。你也可以聲明一個依賴于多個帶有 yield 的依賴,等等。

你可以使用帶有 yield 的依賴項,并且可以包含 try 代碼塊用于捕獲異常。同樣,你可以在 yield 之后的退出代碼中拋出一個 HTTPException 或類似的異常。

from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPExceptionapp = FastAPI()data = {"plumbus": {"description": "Freshly pickled plumbus", "owner": "Morty"},"portal-gun": {"description": "Gun to create portals", "owner": "Rick"},
}class OwnerError(Exception):passdef get_username():try:yield "Rick"except OwnerError as e:raise HTTPException(status_code=400, detail=f"Owner error: {e}")@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):if item_id not in data:raise HTTPException(status_code=404, detail="Item not found")item = data[item_id]if item["owner"] != username:raise OwnerError(username)return item

如果你在包含 yield 的依賴項中使用 except 捕獲了一個異常,然后你沒有重新拋出該異常(或拋出一個新異常),與在普通的Python代碼中相同,FastAPI不會注意到發生了異常。

如果你在使用 yield 的依賴項中捕獲到了一個異常,你應該再次拋出捕獲到的異常,除非你拋出 HTTPException 或類似的其他異常,你可以使用 raise 再次拋出捕獲到的異常。

高級依賴項

我們之前看到的所有依賴項都是寫死的函數或類,但也可以為依賴項設置參數,避免聲明多個不同的函數或類。

Python 可以把類實例變為可調用項。這里說的不是類本身(類本就是可調用項),而是類實例。

為此,需要聲明 __call__ 方法,對象加括號再次調用走的是__call__方法!

from fastapi import Depends, FastAPIapp = FastAPI()class FixedContentQueryChecker:def __init__(self, fixed_content: str):self.fixed_content = fixed_contentdef __call__(self, q: str = ""):if q:return self.fixed_content in qreturn Falsechecker = FixedContentQueryChecker("bar")@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):return {"fixed_content_in_query": fixed_content_included}

不要再在 Depends(checker) 中使用 Depends(FixedContentQueryChecker), 而是要使用 checker,因為依賴項是類實例 - checker,不是類。

處理依賴項時,FastAPI 以如下方式調用 checker:

checker(q="somequery")

并用路徑操作函數的參數 fixed_content_included 返回依賴項的值。

子路由APIRouter

和Flask的藍圖類型,APIRouter也可以視為一個「迷你 FastAPI」類。

from fastapi import APIRouter, Depends, HTTPExceptionfrom ..dependencies import get_token_headerrouter = APIRouter(prefix="/items",tags=["items"],dependencies=[Depends(get_token_header)],responses={404: {"description": "Not found"}},
)fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}@router.get("/")
async def read_items():return fake_items_db@router.get("/{item_id}")
async def read_item(item_id: str):if item_id not in fake_items_db:raise HTTPException(status_code=404, detail="Item not found")return {"name": fake_items_db[item_id]["name"], "item_id": item_id}@router.put("/{item_id}",tags=["custom"],responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):if item_id != "plumbus":raise HTTPException(status_code=403, detail="You can only update the item: plumbus")return {"item_id": item_id, "name": "The great Plumbus"}
from fastapi import Depends, FastAPIfrom .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, usersapp = FastAPI(dependencies=[Depends(get_query_token)])app.include_router(users.router)
app.include_router(items.router)
app.include_router(admin.router,prefix="/admin",tags=["admin"],dependencies=[Depends(get_token_header)],responses={418: {"description": "I'm a teapot"}},
)@app.get("/")
async def root():return {"message": "Hello Bigger Applications!"}

為什么這樣配置?

FastAPI 的路徑操作函數中,參數不僅僅用于傳遞請求數據(如 URL 參數、查詢參數、請求體),還可以傳遞很多其他內容,比如:

  • 請求參數

    • Path, Query, Header, Cookie:顯式標記參數來自哪里;
    • 你甚至可以用 Pydantic 模型接收復雜結構(如 JSON 請求體);
  • 依賴注入

    • 使用 Depends() 傳入函數或類,做認證、權限控制、數據庫連接等;
  • 請求對象

    • Request, Response, BackgroundTasks, UploadFile, Form 等可以直接作為參數傳入;
  • 響應控制

    • response_model, status_code, responses 通過裝飾器或參數控制返回值格式和狀態碼;
  • 中間層邏輯

    • 可以將各種邏輯從主業務函數中解耦,寫到參數函數中再注入進來。

使用者會感覺,除了業務邏輯,所有功能都“被塞”進了函數參數 —— 這不是錯覺,這是 FastAPI 的設計哲學。

FastAPI 的核心設計理念可以總結為以下幾點:

? 類型提示 + 自動解析:利用 Python 3.6+ 的類型提示(type hinting),FastAPI 能夠自動:

  • 校驗參數類型(如 int, str | None, list[str]
  • 生成交互式文檔
  • 明確函數簽名和文檔語義

這就意味著你寫的每一個參數,既是校驗規則,也是接口說明文檔,一舉多得。

? 顯式優于隱式(Pythonic):傳統框架中很多功能依賴隱式機制,比如:

  • Flask 用全局 request 對象(基于 threading.local);
  • Django 用中間件傳遞認證信息;

而 FastAPI 強調顯式聲明:

async def get_current_user(user: User = Depends(authenticate_user)):

誰來、從哪兒來、做什么,都明明白白列出來。

FastAPI 還強調“組合式編程”(Composable Programming):

  • 認證邏輯、數據庫連接、緩存控制、分頁器等通用邏輯都可以通過參數注入的方式分離出來;
  • 主業務函數只關注業務,其他邏輯通過參數“拼裝”進來,提升復用性和測試性。

為什么不采用“傳統的請求-響應模型”?

傳統模型(如 Flask)通常是:

from flask import request@app.route("/user")
def get_user():user_id = request.args.get("id")return {"user_id": user_id}

它有幾個問題:

  • 參數缺乏類型校驗;
  • 請求來源混亂(你不知道 id 是 query 還是 path);
  • 邏輯與框架綁定緊密(全局對象如 request 不易測試);
  • 很難自動生成高質量的 API 文檔。

FastAPI 的方式雖然一開始看著“重”,但具備:

  • 更強的結構化
  • 更好的 IDE 補全和類型檢查
  • 自動文檔
  • 解耦邏輯的能力
  • 明確的請求來源

說到底,FastApi的目標是構建 可維護、自動文檔化、類型安全、可測試的 Web API,而不是一個高可用的Web API。

中間件

你可以向 FastAPI 應用添加中間件,"中間件"是一個函數,它在每個請求被特定的路徑操作處理之前,以及在每個響應返回之前工作.

  • 它接收你的應用程序的每一個請求.
  • 然后它可以對這個請求做一些事情或者執行任何需要的代碼.
  • 然后它將請求傳遞給應用程序的其他部分 (通過某種路徑操作).
  • 然后它獲取應用程序生產的響應 (通過某種路徑操作).
  • 它可以對該響應做些什么或者執行任何需要的代碼.
  • 然后它返回這個 響應.

如果你使用了 yield 關鍵字依賴, 依賴中的退出代碼將在執行中間件后執行,可以將yield理解為go的defer。

要創建中間件,你可以在函數的頂部使用裝飾器 @app.middleware("http")

中間件的參數包括:

  • request:傳入的請求對象;
  • 一個函數 call_next,它接收 request 作為參數;
    • call_next(request) 會將請求傳遞給對應的路徑操作函數;
    • 然后返回由路徑操作生成的 response
  • 你可以在返回 response 之前對其進一步修改。
import timefrom fastapi import FastAPI, Requestapp = FastAPI()@app.middleware("http")
async def add_process_time_header(request: Request, call_next):start_time = time.perf_counter()response = await call_next(request)process_time = time.perf_counter() - start_timeresponse.headers["X-Process-Time"] = str(process_time)return response

CORS(跨域資源共享)

CORS 或者「跨域資源共享」 指瀏覽器中運行的前端擁有與后端通信的 JavaScript 代碼,而后端處于與前端不同的「源」的情況。

源是協議(http、https)、域(myapp.com、localhost、localhost.tiangolo.com)以及端口(80、443、8080)的組合。

因此,以下都是不同的源

  • http://localhost
  • https://localhost
  • http://localhost:8080

即使它們都在 localhost 中,但由于使用了不同的協議或端口,它們仍然被視為不同的「源」。

假設你的瀏覽器中有一個前端運行在 http://localhost:8080,并且它的 JavaScript 正在嘗試與運行在 http://localhost 的后端通信(因為我們沒有指定端口,瀏覽器會采用默認的端口 80)。

然后,瀏覽器會向后端發送一個 HTTP OPTIONS 請求。如果后端發送了適當的 headers 來授權來自這個不同源(http://localhost:8080)的通信,瀏覽器將允許前端的 JavaScript 向后端發送請求。

為此,后端必須有一個「允許的源」列表,在這種情況下,它必須包含 http://localhost:8080,前端才能正常工作。

可以使用 "*"(一個「通配符」)聲明這個列表,表示全部都是允許的。

但這僅允許某些類型的通信,不包括所有涉及憑據的內容,如 Cookies 以及那些使用 Bearer 令牌的授權 headers 等。

因此,為了一切都能正常工作,最好顯式地指定允許的源,你可以在 FastAPI 應用中使用 CORSMiddleware 來配置它。

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddlewareapp = FastAPI()origins = ["http://localhost.tiangolo.com","https://localhost.tiangolo.com","http://localhost","http://localhost:8080",
]app.add_middleware(CORSMiddleware,allow_origins=origins,allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)@app.get("/")
async def main():return {"message": "Hello World"}

默認情況下,這個 CORSMiddleware 實現所使用的默認參數較為保守,所以你需要顯式地啟用特定的源、方法或 headers,以便瀏覽器能夠在跨域上下文中使用它們。

支持的參數如下:

  • allow_origins
    一個允許跨域請求的源列表。例如:
    ['https://example.org', 'https://www.example.org']
    你也可以使用 ['*'] 來允許任何源。

  • allow_origin_regex
    一個正則表達式字符串,匹配的源將允許跨域請求。
    例如:'https://.*\.example\.org'

  • allow_methods
    一個允許跨域請求的 HTTP 方法列表。默認為 ['GET']
    可以使用 ['*'] 來允許所有標準方法(如 POSTPUTDELETE 等)。

  • allow_headers
    一個允許跨域請求的 HTTP 請求頭列表。默認為 []
    使用 ['*'] 可以允許所有的請求頭。
    其中 AcceptAccept-LanguageContent-LanguageContent-Type 總是被允許的。

  • allow_credentials
    是否允許攜帶 cookies(憑證)。默認為 False
    注意:若設置為 True,則 allow_origins 不能為 ['*'],必須明確指定源。

  • expose_headers
    指示哪些響應頭可以被瀏覽器訪問。默認為 []

  • max_age
    指定瀏覽器緩存預檢請求(CORS 響應)的最長時間(單位:秒)。默認為 600

CORSMiddleware 主要響應兩種特定類型的 HTTP 請求:

  1. 預檢請求(OPTIONS):瀏覽器在實際請求前自動發起,用于驗證是否允許跨域;
  2. 實際跨域請求:在通過預檢后,才允許真正的 API 請求被發送。

其他中間件

FastAPI(實際上是 Starlette)提供了一種更簡單的方式,能讓內部中間件在處理服務器錯誤的同時,還能讓自定義異常處理器正常運作。app.add_middleware() 的第一個參數是中間件的類,其它參數則是要傳遞給中間件的參數。

HTTPSRedirectMiddleware

HTTPSRedirectMiddleware 強制所有傳入請求必須是 httpswss
任何傳向 httpws 的請求都會被自動重定向至安全方案。

示例:

from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddlewareapp = FastAPI()
app.add_middleware(HTTPSRedirectMiddleware)@app.get("/")
async def main():return {"message": "Hello World"}

TrustedHostMiddleware

TrustedHostMiddleware 強制所有傳入請求必須正確設置 Host 請求頭,以防止 HTTP 主機頭攻擊。

示例:

from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddlewareapp = FastAPI()
app.add_middleware(TrustedHostMiddleware,allowed_hosts=["example.com", "*.example.com"]
)@app.get("/")
async def main():return {"message": "Hello World"}

支持參數:

  • allowed_hosts
    允許的域名(主機名)列表。例如:*.example.com 可匹配任意子域名。
    也可使用 ["*"] 允許任意主機名,或不添加中間件跳過驗證。
    如果傳入請求未通過驗證,將返回 400 響應。

GZipMiddleware

GZipMiddleware 用于在客戶端的 Accept-Encoding 請求頭包含 gzip 時,自動返回 GZip 壓縮響應。
它可處理標準響應與流響應。

示例:

from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddlewareapp = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=5)@app.get("/")
async def main():return "somebigcontent"

支持參數:

  • minimum_size
    響應內容小于該字節數時不進行 GZip 壓縮。默認值為 500 字節。

  • compresslevel
    GZip 壓縮級別,取值范圍為 0-9,默認為 9(壓縮率最高,速度最慢)。

和依賴項的區別

特性中間件(Middleware)依賴項(Dependencies)
作用層級應用于整個應用的請求/響應生命周期,處理所有請求作用于特定路徑操作函數或多個函數,支持參數注入
執行時機請求進入 FastAPI 后最先執行,響應返回前最后執行在路徑操作函數執行前運行,支持參數準備、驗證、注入
功能范圍通常用于處理跨域、認證、日志、限流、壓縮等全局功能用于業務邏輯中參數處理、復用代碼、復雜校驗、依賴注入
返回值返回修改后的響應(Response)返回參數值,作為路徑操作函數參數
定義方式使用 @app.middleware("http") 裝飾函數或 add_middleware()使用函數或類,并通過 Depends() 注入
設計目標處理所有請求的公共橫切關注點解耦業務邏輯,復用代碼,提高代碼模塊化
例子日志記錄、中間件認證、請求限流、壓縮響應驗證用戶身份、獲取數據庫連接、復用公共參數

為什么 FastAPI 分離中間件和依賴? FastAPI 的設計追求高度的靈活性和清晰的職責分離:

  • 中間件 設計為應用級的請求/響應處理鏈,用于統一處理所有請求,不依賴具體業務邏輯。
  • 依賴項 設計為路徑操作函數的參數注入機制,便于將業務相關的功能按需引入,支持復雜的參數驗證與邏輯復用。

這種設計讓開發者可以:

  • 在不修改業務代碼的情況下,統一處理請求和響應(通過中間件);
  • 在業務代碼中靈活注入所需資源或邏輯(通過依賴),保持代碼簡潔且模塊化。

后臺任務

你可以定義在返回響應后運行的后臺任務,這對需要在請求之后執行的操作很有用,但客戶端不必在接收響應之前等待操作完成。

首先導入 BackgroundTasks 并在 路徑操作函數 中使用類型聲明 BackgroundTasks 定義一個參數:

from fastapi import BackgroundTasks, FastAPIapp = FastAPI()def write_notification(email: str, message=""):with open("log.txt", mode="w") as email_file:content = f"notification for {email}: {message}"email_file.write(content)@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):background_tasks.add_task(write_notification, email, message="some notification")return {"message": "Notification sent in the background"}

FastAPI 會創建一個 BackgroundTasks 類型的對象并作為該參數傳入,在你的 路徑操作函數 里,用 .add_task() 方法將任務函數傳到 后臺任務 對象中!

.add_task() 接收以下參數:

  • 在后臺運行的任務函數(write_notification)。
  • 應按順序傳遞給任務函數的任意參數序列(email)。
  • 應傳遞給任務函數的任意關鍵字參數(message=“some notification”)。

使用 BackgroundTasks 也適用于依賴注入系統,你可以在多個級別聲明 BackgroundTasks 類型的參數:在 路徑操作函數 里,在依賴中(可依賴),在子依賴中,等等。

FastAPI 知道在每種情況下該做什么以及如何復用同一對象,因此所有后臺任務被合并在一起并且隨后在后臺運行:

from typing import Annotatedfrom fastapi import BackgroundTasks, Depends, FastAPIapp = FastAPI()def write_log(message: str):with open("log.txt", mode="a") as log:log.write(message)def get_query(background_tasks: BackgroundTasks, q: str | None = None):if q:message = f"found query: {q}\n"background_tasks.add_task(write_log, message)return q@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):message = f"message to {email}\n"background_tasks.add_task(write_log, message)return {"message": "Message sent"}

實際上就是用線程池去執行這些任務,只不過是這個線程池是基于threading + anyio + asyncio的牛逼的線程池,比我們自己創建的pool = ThreadingPool()高級一些。

生命周期事件

你可以定義在應用啟動前執行的邏輯(代碼)。這意味著在應用開始接收請求之前,這些代碼只會被執行一次。

同樣地,你可以定義在應用關閉時應執行的邏輯。在這種情況下,這段代碼將在處理可能的多次請求后執行一次。

因為這段代碼在應用開始接收請求之前執行,也會在處理可能的若干請求之后執行,它覆蓋了整個應用程序的生命周期("生命周期"這個詞很重要😉)。

假設你有幾個機器學習的模型,你想要用它們來處理請求。相同的模型在請求之間是共享的,因此并非每個請求或每個用戶各自擁有一個模型。假設加載模型可能需要相當長的時間,因為它必須從磁盤讀取大量數據。因此你不希望每個請求都加載它。

你可以在模塊/文件的頂部加載它,但這也意味著即使你只是在運行一個簡單的自動化測試,它也會加載模型,這樣測試將變慢,因為它必須在能夠獨立運行代碼的其他部分之前等待模型加載完成。

這就是我們要解決的問題——在處理請求前加載模型,但只是在應用開始接收請求前,而不是代碼執行時。

你可以使用FastAPI()應用的lifespan參數和一個上下文管理器來定義啟動和關閉的邏輯,FastAPI() 的 lifespan 參數接受一個異步上下文管理器,所以我們可以把我們定義的上下文管理器 lifespan 傳給它。

from contextlib import asynccontextmanagerfrom fastapi import FastAPIdef fake_answer_to_everything_ml_model(x: float):return x * 42ml_models = {}@asynccontextmanager
async def lifespan(app: FastAPI):# Load the ML modelml_models["answer_to_everything"] = fake_answer_to_everything_ml_modelyield# Clean up the ML models and release the resourcesml_models.clear()app = FastAPI(lifespan=lifespan)@app.get("/predict")
async def predict(x: float):result = ml_models["answer_to_everything"](x)return {"result": result}

你可以像上面一樣創建了一個上下文管理器或者異步上下文管理器,它的作用是在進入 with 塊時,執行 yield 之前的代碼,并且在離開 with 塊時,執行 yield 后面的代碼。

配置啟動和關閉事件的推薦方法是使用 FastAPI() 應用的 lifespan 參數,如前所示。如果你提供了一個 lifespan 參數,啟動(startup)和關閉(shutdown)事件處理器將不再生效。要么使用 lifespan,要么配置所有事件,兩者不能共用。

使用 startup 事件聲明 app 啟動前運行的函數:

from fastapi import FastAPIapp = FastAPI()items = {}@app.on_event("startup")
async def startup_event():items["foo"] = {"name": "Fighters"}items["bar"] = {"name": "Tenders"}@app.get("/items/{item_id}")
async def read_items(item_id: str):return items[item_id]

使用 shutdown 事件聲明 app 關閉時運行的函數:

from fastapi import FastAPIapp = FastAPI()@app.on_event("shutdown")
def shutdown_event():with open("log.txt", mode="a") as log:log.write("Application shutdown")@app.get("/items/")
async def read_items():return [{"name": "Foo"}]

啟動和關閉的邏輯很可能是連接在一起的,你可能希望啟動某個東西然后結束它,獲取一個資源然后釋放它等等。

在不共享邏輯或變量的不同函數中處理這些邏輯比較困難,因為你需要在全局變量中存儲值或使用類似的方式。

SQL(關系型)數據庫——現代化ORM

FastAPI 并不要求您使用 SQL(關系型)數據庫。您可以使用任何想用的數據庫。

SQLModel 是基于 SQLAlchemy 和 Pydantic 構建的。它由 FastAPI 的同一作者制作,旨在完美匹配需要使用 SQL 數據庫的 FastAPI 應用程序。

由于 SQLModel 基于 SQLAlchemy,因此您可以輕松使用任何由 SQLAlchemy 支持的數據庫(這也讓它們被 SQLModel 支持),例如:

  • PostgreSQL
  • MySQL
  • SQLite
  • Oracle
  • Microsoft SQL Server 等.

安裝SQLModel:pip install sqlmodel

導入 SQLModel 并創建一個數據庫模型:

from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, selectclass Hero(SQLModel, table=True):id: int | None = Field(default=None, primary_key=True)name: str = Field(index=True)age: int | None = Field(default=None, index=True)secret_name: str# Code below omitted 👇

Hero 類與 Pydantic 模型非常相似(實際上,從底層來看,它確實就是一個 Pydantic 模型)。

有一些區別:

  • table=True 會告訴 SQLModel 這是一個表模型,它應該表示 SQL 數據庫中的一個表,而不僅僅是一個數據模型(就像其他常規的 Pydantic 類一樣)。

  • Field(primary_key=True) 會告訴 SQLModel id 是 SQL 數據庫中的主鍵(您可以在 SQLModel 文檔中了解更多關于 SQL 主鍵的信息)。

  • 把類型設置為 int | None,SQLModel 就能知道該列在 SQL 數據庫中應該是 INTEGER 類型,并且應該是 NULLABLE

  • Field(index=True) 會告訴 SQLModel 應該為此列創建一個 SQL 索引,這樣在讀取按此列過濾的數據時,程序能在數據庫中進行更快的查找。

  • SQLModel 會知道聲明為 str 的內容將是類型為 TEXT(或 VARCHAR,具體取決于數據庫)的 SQL 列。

SQLModel 的引擎 engine(實際上它是一個 SQLAlchemy engine )是用來與數據庫保持連接的。您只需構建一個 engine,來讓您的所有代碼連接到同一個數據庫。

# Code above omitted 👆sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)# Code below omitted 👇

使用 check_same_thread=False 可以讓 FastAPI 在不同線程中使用同一個 SQLite 數據庫。這很有必要,因為單個請求可能會使用多個線程(例如在依賴項中)。

然后,我們來添加一個函數,使用 SQLModel.metadata.create_all(engine) 為所有表模型創建表。

# Code above omitted 👆def create_db_and_tables():SQLModel.metadata.create_all(engine)# Code below omitted 👇

Session 會存儲內存中的對象并跟蹤數據中所需更改的內容,然后它使用 engine 與數據庫進行通信。

我們會使用 yield 創建一個 FastAPI 依賴項,為每個請求提供一個新的 Session 。這確保我們每個請求使用一個單獨的會話。

# Code above omitted 👆def get_session():with Session(engine) as session:yield sessionSessionDep = Annotated[Session, Depends(get_session)]# Code below omitted 👇

我們會在應用程序啟動時創建數據庫表。

# Code above omitted 👆app = FastAPI()@app.on_event("startup")
def on_startup():create_db_and_tables()# Code below omitted 👇

這里,我們使用 SessionDep 依賴項(一個 Session )將新的 Hero 添加到 Session 實例中,提交更改到數據庫,刷新 hero 中的數據,并返回它。

# Code above omitted 👆@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep) -> Hero:session.add(hero)session.commit()session.refresh(hero)return hero# Code below omitted 👇

我們可以使用 select() 從數據庫中讀取 Hero 類,并利用 limit 和 offset 來對結果進行分頁。

# Code above omitted 👆@app.get("/heroes/")
def read_heroes(session: SessionDep,offset: int = 0,limit: Annotated[int, Query(le=100)] = 100,
) -> list[Hero]:heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()return heroes# Code below omitted 👇

我們可以讀取單個 Hero 。

# Code above omitted 👆@app.get("/heroes/{hero_id}")
def read_hero(hero_id: int, session: SessionDep) -> Hero:hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")return hero# Code below omitted 👇

我們也可以刪除單個 Hero 。

# Code above omitted 👆@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")session.delete(hero)session.commit()return {"ok": True}

靜態文件

您可以使用 StaticFiles從目錄中自動提供靜態文件。

  • 導入StaticFiles。
  • “掛載”(Mount) 一個 StaticFiles() 實例到一個指定路徑。
from fastapi import FastAPI
from fastapi.staticfiles import StaticFilesapp = FastAPI()app.mount("/static", StaticFiles(directory="static"), name="static")

“掛載” 表示在特定路徑添加一個完全"獨立的"應用,然后負責處理所有子路徑。

這個 “子應用” 會被 “掛載” 到第一個 “/static” 指向的子路徑。因此,任何以"/static"開頭的路徑都會被它處理。

  • directory=“static” 指向包含你的靜態文件的目錄名字。
  • name=“static” 提供了一個能被FastAPI內部使用的名字。

模板

Flask 等工具使用的 Jinja2 是最用的模板引擎。(pip install jinja2)

使用 Jinja2Templates

  • 導入 Jinja2Templates
  • 創建 可復用的 templates 對象
  • 在返回模板的路徑操作中 聲明 Request 參數
  • 使用 templates 渲染并返回 TemplateResponse
    • 傳遞模板的名稱
    • 傳遞 request 對象
    • 傳遞一個包含多個鍵值對(用于 Jinja2 模板)的 "context" 字典
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templatesapp = FastAPI()app.mount("/static", StaticFiles(directory="static"), name="static")templates = Jinja2Templates(directory="templates")@app.get("/items/{id}", response_class=HTMLResponse)
async def read_item(request: Request, id: str):return templates.TemplateResponse(request=request, name="item.html", context={"id": id})

通過聲明 response_class=HTMLResponse,API 文檔就能識別響應的對象是 HTML。

編寫模板 templates/item.html,代碼如下:

<html>
<head><title>Item Details</title><link href="{{ url_for('static', path='/styles.css') }}" rel="stylesheet">
</head>
<body><h1><a href="{{ url_for('read_item', id=id) }}">Item ID: {{ id }}</a></h1>
</body>
</html>

部署

部署 FastAPI 應用程序相對容易。

簡而言之,使用 fastapi run 來運行您的 FastAPI 應用程序:

在這里插入圖片描述
這在大多數情況下都能正常運行。😎

FastAPI 使用了一種用于構建 Python Web 框架和服務器的標準,稱為 ASGI。FastAPI 本質上是一個 ASGI Web 框架

要在遠程服務器上運行 FastAPI 應用(或任何其他 ASGI 應用),您需要一個 ASGI 服務器程序,例如 Uvicorn。它是 fastapi 命令默認使用的 ASGI 服務器。

除此之外,還有其他一些可選的 ASGI 服務器,例如:

  • Uvicorn:高性能 ASGI 服務器。
  • Hypercorn:與 HTTP/2 和 Trio 等兼容的 ASGI 服務器。
  • Daphne:為 Django Channels 構建的 ASGI 服務器。
  • Granian:基于 Rust 的 HTTP 服務器,專為 Python 應用設計。
  • NGINX Unit:輕量級且靈活的 Web 應用運行時環境。

當您安裝 FastAPI 時,它自帶一個生產環境服務器——Uvicorn,并且您可以使用 fastapi run 命令來啟動它。

不過,您也可以手動安裝 ASGI 服務器:pip install "uvicorn[standard]"

如果您手動安裝了 ASGI 服務器,通常需要以特定格式傳遞一個導入字符串,以便服務器能夠正確導入您的 FastAPI 應用:

uvicorn main:app --host 0.0.0.0 --port 80

到目前為止,在文檔中的所有教程中,您可能一直是在運行一個服務器程序,例如使用 fastapi 命令來啟動 Uvicorn,而它默認運行的是單進程模式。

部署應用程序時,您可能希望進行一些進程復制,以利用多核 CPU 并能夠處理更多請求。

您可以使用 --workers 命令行選項來啟動多個工作進程:

在這里插入圖片描述

在這里插入圖片描述
如果您的代碼加載 1 GB 大小的機器學習模型,則當您使用 API 運行一個進程時,它將至少消耗 1 GB RAM。 如果您啟動 4 個進程(4 個工作進程),每個進程將消耗 1 GB RAM。 因此,您的 API 總共將消耗 4 GB RAM。

如果您的遠程服務器或虛擬機只有 3 GB RAM,嘗試加載超過 4 GB RAM 將導致問題。

在這里插入圖片描述

本文來自互聯網用戶投稿,該文觀點僅代表作者本人,不代表本站立場。本站僅提供信息存儲空間服務,不擁有所有權,不承擔相關法律責任。
如若轉載,請注明出處:http://www.pswp.cn/diannao/83588.shtml
繁體地址,請注明出處:http://hk.pswp.cn/diannao/83588.shtml
英文地址,請注明出處:http://en.pswp.cn/diannao/83588.shtml

如若內容造成侵權/違法違規/事實不符,請聯系多彩編程網進行投訴反饋email:809451989@qq.com,一經查實,立即刪除!

相關文章

DAY 4 缺失值的處理

\1. 打開數據 import pandas as pd data pd.read_csv(rdata.csv) data\2. 查看數據 # 打印數據集的基本信息&#xff08;列名、非空值數量、數據類型等&#xff09; print("data.info() - 數據集的基本信息&#xff08;列名、非空值數量、數據類型等&#xff09;&#…

Java面試實戰:從Spring Boot到分布式緩存的深度探索

Java面試實戰&#xff1a;從Spring Boot到分布式緩存的深度探索 場景介紹 在一家著名的互聯網大廠&#xff0c;面試官老王正對求職者“水貨程序員”明哥進行Java技術面試。明哥帶著一點緊張和自信&#xff0c;迎接這場技術“拷問”。 第一輪&#xff1a;基礎問題 老王&#…

UART、SPI、IIC復習總結

一、UART 1、UART和USART的異同&#xff1f; 相同點 基本功能&#xff1a;都是用于串行通信的數據收發設備&#xff0c;能夠實現數據在不同設備之間的傳輸。在異步通信模式下&#xff0c;二者的工作方式相似&#xff0c;都使用起始位、數據位、校驗位&#xff08;可選&#…

PostGIS實現矢量數據轉柵格數據【ST_AsRaster】

ST_AsRaster函數應用詳解&#xff1a;將矢量數據轉換為柵格數據 [文章目錄] 一、函數概述 二、函數參數與分組說明 三、核心特性與注意事項 四、示例代碼 五、應用場景 六、版本依賴 七、總結 一、函數概述 ST_AsRaster是PostGIS中用于將幾何對象&#xff08;如點、線…

Linux 線程(上)

前言&#xff1a;大家早上中午晚上好&#xff01;&#xff01;今天來學習一下linux系統下所謂的線程吧&#xff01;&#xff01;&#xff01; 一、重新理解進程&#xff0c;什么是進程&#xff1f; 1.1 圖解 其中黑色虛線部分一整塊就是進程&#xff0c;注意&#xff1a;一整…

Java API學習筆記

一.類 1. String 類 不可變性&#xff1a;String對象創建后不可修改&#xff0c;每次操作返回新對象 String str "Hello"; str.length(); str.charAt(0); str.substring(1, 4); str.indexOf("l"); str.equals("hel…

醫療信息系統安全防護體系的深度構建與理論實踐融合

一、醫療數據訪問系統的安全挑戰與理論基礎 1.1 系統架構安全需求分析 在醫療信息系統中&#xff0c;基于身份標識的信息查詢功能通常采用分層架構設計&#xff0c;包括表現層、應用層和數據層。根據ISO/IEC 27001信息安全管理體系要求&#xff0c;此類系統需滿足數據保密性…

5.18本日總結

一、英語 復習list3list28 二、數學 學習14講部分內容&#xff0c;1000題13講部分 三、408 學習計網5.3剩余內容 四、總結 計網TCP內容比較重要&#xff0c;連接過程等要時常復習&#xff1b;高數學到二重積分對定積分的計算相關方法有所遺忘&#xff0c;需要加強鞏固。…

MATLAB2025新功能

截至2023年9月&#xff0c;MATLAB官方尚未公布2025版本的具體更新內容。根據歷史更新規律和技術發展趨勢&#xff0c;未來版本可能會在以下方面進行優化&#xff1a; AI與深度學習增強 可能新增自動化模型壓縮工具強化生成式AI模型支持改進的ONNX格式轉換接口 性能提升 矩陣運…

算法題(149):矩陣消除游戲

審題&#xff1a; 本題需要我們找到消除矩陣行與列后可以獲得的最大權值 思路&#xff1a; 方法一&#xff1a;貪心二進制枚舉 這里的矩陣消除時&#xff0c;行與列的消除會互相影響&#xff0c;所以如果我們先統計所有行和列的總和&#xff0c;然后選擇消除最大的那一行/列&am…

Uniapp、Flutter 和 React Native 全面對比

文章目錄 前言Uni-app、Flutter 和 React Native 跨平臺框架對比報告1. 性能對比2. 跨平臺能力3. 學習曲線4. 社區生態與第三方庫5. 原生能力擴展6. UI 渲染能力7. 企業支持與典型使用場景8. 開發效率與工具鏈 前言 將對 Uniapp、Flutter 和 React Native 進行全面對比&#x…

JAVA SE 多線程(上)

文章目錄 &#x1f4d5;1. Thread類及常見方法??1.1 創建線程??1.2 Thread 的常見構造方法??1.3 Thread 的幾個常見屬性??1.4 啟動一個線程---start()??1.5 中斷一個線程---interrupt()??1.6 等待一個線程---join()??1.7 獲取當前線程引用??1.8 休眠當前線程 &…

Linux云計算訓練營筆記day10(MySQL數據庫)

Linux云計算訓練營筆記day10&#xff08;MySQL數據庫&#xff09; 目錄 Linux云計算訓練營筆記day10&#xff08;MySQL數據庫&#xff09;ifnull別名聚合函數group byHAVING 子查詢關聯查詢 ifnull 在DQL語句中可以使用函數或表達式 函數 IFNULL(arg1,arg2) 如果arg1為NULL,函…

上位機知識篇---流式Web服務器模式的實現

文章目錄 前言 前言 本文簡單介紹了流式Web服務器模式的實現。

Dify與n8n全面對比指南:AI應用開發與工作流自動化平臺選擇【2025最新】

Dify與n8n全面對比指南&#xff1a;AI應用開發與工作流自動化平臺選擇【2025最新】 隨著AI技術與自動化工具的迅速發展&#xff0c;開發者和企業面臨著多種平臺選擇。Dify和n8n作為兩個備受關注的自動化平臺&#xff0c;分別專注于不同領域&#xff1a;Dify主要面向AI應用開發&…

day19-線性表(順序表)(鏈表I)

一、補充 安裝軟件命令&#xff1a; sudo apt-get install (軟件名) 安裝格式化對齊&#xff1a;sudo apt-get install clang-format內存泄漏檢測工具&#xff1a; sudo apt-get install valgrind 編譯后&#xff0c;使用命令 valgrind ./a.out 即可看內存是…

AI:人形機器人一定是人的形狀嗎?

本文將從技術角度分析人形機器人是否必須是人的形狀&#xff0c;以及人形與非人形機器人在適用場合、優缺點上的差異。以下是詳細解答&#xff1a; 人形機器人一定是人的形狀嗎&#xff1f; 不&#xff0c;人形機器人&#xff08;Humanoid Robot&#xff09;在技術上通常指外…

布隆過濾器和布谷鳥過濾器

原文鏈接&#xff1a;布隆過濾器和布谷鳥過濾器 布隆過濾器 介紹 布隆過濾器&#xff08;Bloom Filter&#xff09;是 1970 年由布隆提出的。它實際上是一個很長的二進制向量和一系列隨機映射函數&#xff0c;檢查值是“可能在集合中”還是“絕對不在集合中” 空間效率高&a…

無需配置光貓,使用網管交換機配合路由器的IPTV功能實現單線復用

一、背景 弱電箱和電視柜只預留了一根網線&#xff0c;路由器放在電視柜&#xff0c;想實現既可以上網又可以正常觀看iptv&#xff0c;本文提供了一種方法。 二、準備工作 1、帶iptv功能的路由器&#xff1b;2、水星sg105pro網管交換機&#xff1b;3、網線若干&#xff1b; …

深入理解SpringBoot中的SpringCache緩存技術

深入理解SpringBoot中的SpringCache緩存技術 引言 在現代應用開發中&#xff0c;緩存技術是提升系統性能的重要手段之一。SpringBoot提供了SpringCache作為緩存抽象層&#xff0c;簡化了緩存的使用和管理。本文將深入探討SpringCache的核心技術點及其在實際業務中的應用場景。…