生命週期事件¶
您可以定義在應用程式啟動之前執行的邏輯（程式碼）。這表示此程式碼將在應用程式開始接收請求之前執行一次。
同樣地，您可以定義在應用程式關閉時執行的邏輯（程式碼）。在這種情況下，此程式碼將在處理完可能許多請求之後執行一次。
因為此程式碼在應用程式開始接收請求之前執行，並且在應用程式完成處理請求之後立即執行，所以它涵蓋了整個應用程式的生命週期（「生命週期」這個詞等一下會很重要😉）。
這對於設定您需要在整個應用程式中使用的資源非常有用，這些資源在請求之間共享，並且/或者您需要在之後清理它們。例如，資料庫連線池，或載入共享的機器學習模型。
使用案例¶
讓我們先從一個使用案例範例開始，然後看看如何用這個解決它。
假設您有一些要用来處理請求的機器學習模型。🤖
相同的模型在請求之間共享，因此，它不是每個請求一個模型，也不是每個使用者一個模型或類似的情況。
假設載入模型可能需要相當長的時間，因為它必須從磁碟讀取大量資料。所以您不想在每個請求都這樣做。
您可以將它載入到模組/檔案的頂層，但這也意味著即使您只是執行一個簡單的自動化測試，它也會載入模型，那麼該測試會很慢，因為它必須等待模型載入才能執行程式碼的獨立部分。
這就是我們要解決的問題，讓我們在處理請求之前載入模型，但僅在應用程式開始接收請求之前，而不是在載入程式碼時。
生命週期¶
您可以使用 FastAPI 應用程式的 lifespan 參數和一個「上下文管理器」（我等一下會告訴您那是什麼）來定義這個啟動和關閉邏輯。
讓我們先從一個例子開始，然後詳細說明。
我們使用 yield 建立一個非同步函式 lifespan() 如下：
from contextlib import asynccontextmanager
from fastapi import FastAPI
def fake_answer_to_everything_ml_model(x: float):
return x * 42
ml_models = {}
@asynccontextmanager
async def lifespan(app: FastAPI):
# Load the ML model
ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model
yield
# Clean up the ML models and release the resources
ml_models.clear()
app = FastAPI(lifespan=lifespan)
@app.get("/predict")
async def predict(x: float):
result = ml_models["answer_to_everything"](x)
return {"result": result}
在這裡，我們通過在 yield 之前將（假的）模型函式放入機器學習模型的字典中來模擬載入模型的昂貴啟動操作。這段程式碼將在應用程式開始接收請求之前，在啟動期間執行。
然後，在 yield 之後，我們卸載模型。這段程式碼將在應用程式完成處理請求之後，在關閉之前執行。例如，這可以釋放記憶體或 GPU 等資源。
提示
當您停止應用程式時，會發生 shutdown。
也許您需要啟動一個新版本，或者您只是厭倦了執行它。🤷
生命週期函式¶
首先要注意的是，我們正在使用 yield 定義一個非同步函式。這與帶有 yield 的依賴項非常相似。
from contextlib import asynccontextmanager
from fastapi import FastAPI
def fake_answer_to_everything_ml_model(x: float):
return x * 42
ml_models = {}
@asynccontextmanager
async def lifespan(app: FastAPI):
# Load the ML model
ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model
yield
# Clean up the ML models and release the resources
ml_models.clear()
app = FastAPI(lifespan=lifespan)
@app.get("/predict")
async def predict(x: float):
result = ml_models["answer_to_everything"](x)
return {"result": result}
函式的第一部分，在 yield 之前，將在應用程式啟動之前執行。
而 `yield` 後面的部分會在應用程式完成**之後**執行。
非同步上下文管理器¶
如果您檢查一下，會發現該函式使用了 `@asynccontextmanager` 裝飾器。
這會將函式轉換為稱為「**非同步上下文管理器**」的東西。
from contextlib import asynccontextmanager
from fastapi import FastAPI
def fake_answer_to_everything_ml_model(x: float):
return x * 42
ml_models = {}
@asynccontextmanager
async def lifespan(app: FastAPI):
# Load the ML model
ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model
yield
# Clean up the ML models and release the resources
ml_models.clear()
app = FastAPI(lifespan=lifespan)
@app.get("/predict")
async def predict(x: float):
result = ml_models["answer_to_everything"](x)
return {"result": result}
Python 中的**上下文管理器**是可以搭配 `with` 陳述式使用的東西，例如，`open()` 可以用作上下文管理器。
with open("file.txt") as file:
file.read()
在最新版本的 Python 中，也有**非同步上下文管理器**。您可以將其與 `async with` 一起使用。
async with lifespan(app):
await do_stuff()
當您像上面那樣創建上下文管理器或非同步上下文管理器時，它的作用是，在進入 `with` 區塊之前，它會執行 `yield` 之前的程式碼，而在離開 `with` 區塊之後，它會執行 `yield` 之後的程式碼。
在我們上面的程式碼範例中，我們沒有直接使用它，而是將其傳遞給 FastAPI 供其使用。
`FastAPI` 應用程式的 `lifespan` 參數接受一個**非同步上下文管理器**，因此我們可以將新的 `lifespan` 非同步上下文管理器傳遞給它。
from contextlib import asynccontextmanager
from fastapi import FastAPI
def fake_answer_to_everything_ml_model(x: float):
return x * 42
ml_models = {}
@asynccontextmanager
async def lifespan(app: FastAPI):
# Load the ML model
ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model
yield
# Clean up the ML models and release the resources
ml_models.clear()
app = FastAPI(lifespan=lifespan)
@app.get("/predict")
async def predict(x: float):
result = ml_models["answer_to_everything"](x)
return {"result": result}
替代事件（已棄用）¶
警告
處理*啟動*和*關閉*的建議方法是使用如上所述的 `FastAPI` 應用程式的 `lifespan` 參數。如果您提供 `lifespan` 參數，則將不再呼叫 `startup` 和 `shutdown` 事件處理程式。它要么全部使用 `lifespan`，要么全部使用事件，不能兩者兼用。
您可以跳過此部分。
還有另一種方法可以定義在*啟動*和*關閉*期間執行的邏輯。
您可以定義需要在應用程式啟動之前或應用程式關閉時執行的事件處理程式（函式）。
這些函式可以用 `async def` 或普通的 `def` 宣告。
`startup` 事件¶
要新增一個應在應用程式啟動之前執行的函式，請使用事件 `"startup"` 宣告它。
from fastapi import FastAPI
app = 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]
在這種情況下，`startup` 事件處理程式函式將使用一些值初始化項目「資料庫」（只是一個 `dict`）。
您可以新增多個事件處理程式函式。
而且在所有 `startup` 事件處理程式完成之前，您的應用程式不會開始接收請求。
`shutdown` 事件¶
要新增一個應在應用程式關閉時執行的函式，請使用事件 `"shutdown"` 宣告它。
from fastapi import FastAPI
app = 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"}]
在這裡，`shutdown` 事件處理程式函式會將文字行 `"Application shutdown"` 寫入檔案 `log.txt`。
資訊
在 `open()` 函式中，`mode="a"` 表示「附加」，因此，該行將被新增到該檔案中任何內容的後面，而不會覆蓋先前的內容。
提示
請注意，在這種情況下，我們使用的是與檔案互動的標準 Python `open()` 函式。
因此，它涉及 I/O（輸入/輸出），這需要「等待」將內容寫入磁碟。
但 `open()` 不使用 `async` 和 `await`。
因此，我們使用標準的 `def` 而不是 `async def` 宣告事件處理程式函式。
`startup` 和 `shutdown` 一起使用¶
您的*啟動*和*關閉*邏輯很可能相互關聯，您可能想要啟動某個東西然後完成它，獲取資源然後釋放它，等等。
如果在不共享邏輯或變數的獨立函數中執行此操作，則會更加困難，因為您需要將值儲存在全域變數或使用類似的技巧。
因此，現在建議改用如上所述的 lifespan。
技術細節¶
僅供好奇的技術愛好者參考的技術細節。🤓
在底層，依據 ASGI 技術規範，這是 Lifespan 協議 的一部分，它定義了稱為 startup 和 shutdown 的事件。
子應用程式¶
🚨 請記住，這些 lifespan 事件（啟動和關閉）只會在主應用程式中執行，而不會在 子應用程式 - 掛載 中執行。