Lifespan イベント¶
🌐 AI と人間による翻訳
この翻訳は、人間のガイドに基づいて AI によって作成されました。🤝
原文の意図を取り違えていたり、不自然な表現になっている可能性があります。🤖
AI LLM をより適切に誘導するのを手伝う ことで、この翻訳を改善できます。
アプリケーションが起動する前に一度だけ実行すべきロジック（コード）を定義できます。これは、アプリケーションがリクエストを受け取り始める前に、そのコードが一度だけ実行される、という意味です。
同様に、アプリケーションがシャットダウンするときに実行すべきロジック（コード）も定義できます。この場合、そのコードは、（多くのリクエストを処理した）後に一度だけ実行されます。
このコードは、アプリケーションがリクエストの受け付けを「開始」する前、そして処理を「終了」した直後に実行されるため、アプリケーションの全体の「Lifespan」（この「lifespan」という言葉はすぐ後で重要になります 😉）をカバーします。
これは、アプリ全体で使用し、リクエスト間で「共有」し、かつ後で「クリーンアップ」する必要があるような「リソース」をセットアップするのにとても便利です。たとえば、データベース接続プールや、共有の機械学習モデルの読み込みなどです。
ユースケース¶
まずはユースケースの例から始めて、これをどのように解決するかを見ていきます。
リクエストを処理するために使用したい「機械学習モデル」がいくつかあると想像してください。🤖
同じモデルをリクエスト間で共有するので、リクエストごとやユーザーごとに別々のモデルを使うわけではありません。
モデルの読み込みにはディスクから大量のデータを読む必要があり、かなり時間がかかるかもしれません。したがって、リクエストごとに読み込みたくはありません。
モジュール/ファイルのトップレベルで読み込むこともできますが、その場合は、たとえ簡単な自動テストを実行するだけでも「モデルを読み込む」ことになり、そのモデルの読み込みを待つ必要があるため、独立したコード部分を走らせるだけのテストでも「遅く」なってしまいます。
これを解決しましょう。リクエストを処理する前にモデルを読み込みますが、コードがロードされている最中ではなく、アプリケーションがリクエストの受け付けを開始する直前だけにします。
Lifespan¶
この「起動時」と「シャットダウン時」のロジックは、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}
🤓 Other versions and variants
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 は、アプリケーションを「停止」するときに発生します。
新しいバージョンを開始する必要があるか、単に実行をやめたくなったのかもしれません。🤷
Lifespan 関数¶
まず注目すべきは、yield を使う非同期関数を定義していることです。これは「yield を使う依存関係（Dependencies）」にとてもよく似ています。
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}
🤓 Other versions and variants
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}
🤓 Other versions and variants
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}
🤓 Other versions and variants
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]
🤓 Other versions and variants
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 のイベントハンドラ関数は items の「データベース」（単なる 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"}]
🤓 Other versions and variants
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" は「追加」（append）を意味します。つまり、そのファイルに既にある内容を上書きせず、行が後ろに追記されます。
豆知識
この例では、ファイルを扱う標準の Python 関数 open() を使っています。
そのため、ディスクへの書き込みを「待つ」必要がある I/O（入力/出力）が関わります。
しかし open() 自体は async や await を使いません。
したがって、イベントハンドラ関数は async def ではなく通常の def で宣言しています。
startup と shutdown をまとめて¶
起動時とシャットダウン時のロジックは関連していることが多いです。何かを開始してから終了したい、リソースを獲得してから解放したい、などです.
共有するロジックや変数のない別々の関数でそれを行うのは難しく、グローバル変数などに値を保存する必要が出てきます。
そのため、現在は上で説明したとおり lifespan を使うことが推奨されています。
技術詳細¶
技術が気になる方への細かな詳細です。🤓
内部的には、ASGI の技術仕様において、これは Lifespan プロトコル の一部であり、startup と shutdown というイベントが定義されています。
情報
Starlette の lifespan ハンドラについては、Starlette の Lifespan ドキュメントで詳しく読むことができます。
コードの他の領域で使える lifespan の状態をどのように扱うかも含まれています。
サブアプリケーション¶
🚨 これらの lifespan イベント（startup と shutdown）はメインのアプリケーションに対してのみ実行され、サブアプリケーション - マウント には実行されないことに注意してください。