バックグラウンドタスク¶
レスポンスを返した 後に 実行されるバックグラウンドタスクを定義できます。
これは、リクエスト後に処理を開始する必要があるが、クライアントがレスポンスを受け取る前に処理を終える必要のない操作に役立ちます。
これには、たとえば次のものが含まれます。
- 作業実行後のメール通知:
- メールサーバーへの接続とメールの送信は「遅い」(数秒) 傾向があるため、すぐにレスポンスを返し、バックグラウンドでメール通知ができます。
- データ処理:
- たとえば、時間のかかる処理を必要とするファイル受信時には、「受信済み」(HTTP 202) のレスポンスを返し、バックグラウンドで処理できます。
BackgroundTasks の使用¶
まず初めに、BackgroundTasks をインポートし、BackgroundTasks の型宣言と共に、path operation 関数 のパラメーターを定義します:
from fastapi import BackgroundTasks, FastAPI
app = 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 型のオブジェクトを作成し、そのパラメーターに渡します。
タスク関数の作成¶
バックグラウンドタスクとして実行される関数を作成します。
これは、パラメーターを受け取ることができる単なる標準的な関数です。
これは async def または通常の def 関数であり、FastAPI はこれを正しく処理します。
ここで、タスク関数はファイル書き込みを実行します (メール送信のシミュレーション)。
また、書き込み操作では async と await を使用しないため、通常の def で関数を定義します。
from fastapi import BackgroundTasks, FastAPI
app = 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"}
バックグラウンドタスクの追加¶
path operations 関数 内で、.add_task() メソッドを使用してタスク関数を background tasks オブジェクトに渡します。
from fastapi import BackgroundTasks, FastAPI
app = 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"}
.add_task() は以下の引数を受け取ります:
- バックグラウンドで実行されるタスク関数 (
write_notification)。 - タスク関数に順番に渡す必要のある引数の列 (
email)。 - タスク関数に渡す必要のあるキーワード引数 (
message="some notification")。
依存性注入¶
BackgroundTasks の使用は依存性注入システムでも機能し、様々な階層 (path operations 関数、依存性 (依存可能性)、サブ依存性など) で BackgroundTasks 型のパラメーターを宣言できます。
FastAPI は、それぞれの場合の処理方法と同じオブジェクトの再利用方法を知っているため、すべてのバックグラウンドタスクがマージされ、バックグラウンドで後で実行されます。
from typing import Union
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: Union[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: str = Depends(get_query)
):
message = f"message to {email}\n"
background_tasks.add_task(write_log, message)
return {"message": "Message sent"}
🤓 Other versions and variants
from typing import Annotated
from fastapi import BackgroundTasks, Depends, FastAPI
app = 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"}
from typing import Annotated, Union
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: Union[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"}
from typing import Union
from fastapi import BackgroundTasks, Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: Union[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"}
Tip
Prefer to use the Annotated version if possible.
from fastapi import BackgroundTasks, Depends, FastAPI
app = 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: str = Depends(get_query)
):
message = f"message to {email}\n"
background_tasks.add_task(write_log, message)
return {"message": "Message sent"}
この例では、レスポンスが送信された 後 にメッセージが log.txt ファイルに書き込まれます。
リクエストにクエリがあった場合、バックグラウンドタスクでログに書き込まれます。
そして、path operations 関数 で生成された別のバックグラウンドタスクは、email パスパラメータを使用してメッセージを書き込みます。
技術的な詳細¶
BackgroundTasks クラスは、starlette.backgroundから直接取得されます。
これは、FastAPI に直接インポート/インクルードされるため、fastapi からインポートできる上に、starlette.backgroundから別の BackgroundTask (末尾に s がない) を誤ってインポートすることを回避できます。
BackgroundTasksのみを使用することで (BackgroundTask ではなく)、Request オブジェクトを直接使用する場合と同様に、それを path operations 関数 パラメーターとして使用し、FastAPI に残りの処理を任せることができます。
それでも、FastAPI で BackgroundTask を単独で使用することは可能ですが、コード内でオブジェクトを作成し、それを含むStarlette Response を返す必要があります。
詳細については、バックグラウンドタスクに関する Starlette の公式ドキュメントを参照して下さい。
警告¶
大量のバックグラウンド計算が必要であり、必ずしも同じプロセスで実行する必要がない場合 (たとえば、メモリや変数などを共有する必要がない場合)、Celery のようなより大きな他のツールを使用するとメリットがあるかもしれません。
これらは、より複雑な構成、RabbitMQ や Redis などのメッセージ/ジョブキューマネージャーを必要とする傾向がありますが、複数のプロセス、特に複数のサーバーでバックグラウンドタスクを実行できます。
ただし、同じ FastAPI アプリから変数とオブジェクトにアクセスする必要がある場合、または小さなバックグラウンドタスク (電子メール通知の送信など) を実行する必要がある場合は、単に BackgroundTasks を使用できます。
まとめ¶
BackgroundTasks をインポートして、path operations 関数 や依存関係のパラメータに BackgroundTasksを使用し、バックグラウンドタスクを追加して下さい。