はじめに

FastAPIで、レスポンスを返した後に特定の処理を実行する仕組みがないかなと調べているときにBackgroundTasks機能を知ったので紹介します。

BackgroundTasksとは

公式サイトを引用します。

レスポンスを返した後に実行されるバックグラウンドタスクを定義できます。

これは、リクエスト後に処理を開始する必要があるが、クライアントがレスポンスを受け取る前に処理を終える必要のない操作に役立ちます。

たとえば、以下のようなケースで使用できます。
BackgroundTasksを使用することで、レスポンスを早く返して非同期で各処理を実行できます。

ログの記録
- APIを呼び出したユーザーのアクセスログをデータベースやファイルに記録
メールの送信
- フォームを送信した際にサーバ側でメール送信
定期的なデータの更新
- API呼び出したユーザーのキャッシュデータをバックグラウンドで更新
大きなファイルの処理
- すぐにレスポンスを返した後でバックグラウンドで圧縮、解析

具体例

公式サイトに載っているサンプルプログラムを見ていきます。

.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"}

これで、send_notificationのレスポンスが返った後でwrite_notification関数が実行されます。

send_notification関数に渡す引数はemailだけで良いです。

background_tasksに引数を渡す必要はありません
リクエストが呼ばれたときにFastAPI側がBackgroundTasksのインスタンスを自動的に返してくれるためです

非常にシンプルです。

終わりに

本記事はほとんど公式サイトの引用ですが、send_notification関数を呼び出すときにbackground_tasksに引数として何を渡せばいいかは公式サイトに書かれておらずよくわからなかったため、誰かの参考になればと思い記事を作成しました。