技術的な話

FastAPIのBackgroundTasksを使用して重い処理をバックグラウンドで実行しつつタスクの状態を取得したい

今回やりたいこと

API経由でバッチ処理とか重い処理を動かしておいて、UI側で状態を監視したい…なんてことあるよね。今回はそんな時の処理を書いていこうと思います。

  • APIからバックグラウンドで重い処理を起動したい
  • バックグラウンド処理の状態を取得して画面側で表示したい
  • 処理中でもAPIへのリクエストは止めたくない

今回はとりあえずAPI処理だけに留めようと思います。
APIが出来ちゃえば画面側はご自由にといった感じです。

環境

  • Python 3.9.2
  • FastAPI 0.68.1

各処理を書いていく

FastAPIにはバックグラウンド処理用にBackground Tasksが用意されています。
公式を見ても分かる通りかなりシンプルに書けます。add_taskにバックグラウンド処理を入れるだけ。後はお任せ。素晴らしい。

https://fastapi.tiangolo.com/tutorial/background-tasks/

バックグラウンド処理

まずはバックグラウンド処理を適当に書いていきます。
処理の代わりに10秒スリープするようにしています。

#!/usr/bin/env python3
import time

class HeavyTask():
    
    wait_time:int = 10
    status:bool = False

    def __init__(self, wait_time:int=10) -> None:
        self.wait_time = wait_time
        print(f'background task initialized. set wait time {self.wait_time} s')
    
    def __call__(self) -> None:
        try:
            self.status = True
            print(f'sleep start.')
            time.sleep(self.wait_time)
        except Exception as e:
            print(e)
        finally:
            self.status = False
        print('sleep finished.')
    
    def get_status(self) -> bool:
        return self.status

ポイントとしては__call__部分でしょうか。add_taskにこのタスクを入れると__call__が呼ばれ処理が実行されます。

状態は今回は処理中・終了を返すようにstatusを使用しています。FastAPI側からはget_statusで状態を取得できるようにしています。
他にも処理結果やエラーコードなんかを入れておけば、色々な使い方が出来そうです。

API側処理

getでは状態を返すようにして、postではバックグラウンド処理を実行をするようにしています。
多重実行とかその辺は考えられていないので悪しからず。

#!/usr/bin/env python3

from modules.heavy_task import HeavyTask
from fastapi import FastAPI, BackgroundTasks
from starlette.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=['*'],
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*']
)

task:HeavyTask = HeavyTask(10)

@app.get("/api/background-task")
def fetch_background_task():
    return {"status": task.get_status()}

@app.post("/api/background-task")
def execute_background_task(background_tasks: BackgroundTasks):
    try:
        background_tasks.add_task(task)
    except Exception as e:
        print(e)
        return {"message": f"error occured. reason: {e}"}
    return {"message": "ok"}

API側では重い処理を作成しておいて、実行タイミングでadd_taskしているのみです。
うーんシンプル。

実行してみる

FastAPIの実行については下記を参照してください。

# reloadオプションでソースコードに変更があった場合に自動的に再読み込みします
$ uvicorn server:app --host 0.0.0.0 --port 3000 --reload

swaggerから実行してみます。
postでバックグラウンド処理実行後に何度かgetを呼び出して状態を取得します。

実行した結果が下記になります。
上手いことバックグラウンド処理実行後はstatusがTrueとなり、終了後はFalseとなります。

まとめ

FastAPIで重い処理をバックグラウンドで実行しつつ状態を取得出来るようにするのは比較的簡単です。
何かしら普段からバッチ処理を実行している人は、FastAPIでAPI作っておいて誰かに叩いてもらう…みたいなことも可能になるかと思います。
画面まで作れないといった人は最悪swagger経由で実行してもらえれば。CUIだと抵抗がある人もこれなら動かせるはず。

-技術的な話
-, ,