はじめに
データアナリティクス事業本部のkobayashiです。
コンテナ環境でFastAPIを使ったプロダクト開発を行っている際にある処理をスケジュールに基づき定期実行する必要がありました。その様な場合には処理専用のエンドポイントを用意してcurlなどで定期的に叩く方法もありますがもっと直接的にPythonのライブラリでジョブ実行のスケジュール管理を方法もあります。今回はPythonのライブラリであるAPSchedulerを使って定期実行を試してみたのでその内容をまとめます。
APSchedulerとは
Python関数を定期的に実行するようにスケジューリング管理を行えるPythonライブラリです。実行するジョブ（Python関数）は簡単に新しいジョブを追加したり削除したりすることができジョブのメンテナンスも簡単です。またスケジュールを外部DBに保存することでスケジューラの再起動をおこなってもスケジュールとジョブ実行状態が保持されます。
There are third party solutions for integrating APScheduler with other frameworksにはDjango、Flaskだけの記載でしたが、今回FastAPIでも動作は問題ありませんでした。
詳しくは以下をご確認ください。
APSchedulerを使ってみる
では早速APSchedulerを使ってみたいと思います。試すこととしてはFastAPIのユーザーガイドのチュートリアル-最初のステップ にある。以下のシンプルなコードをベースにAPSchedulerを使って定期実行する処理を追加していきます。
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}環境
- Python: 3.11.4
- APScheduler: 3.10.1
インストール
インストールはいつも通りpipで簡単に済ませます。
$ pip install apschedulerFastAPIでAPSchedulerを使ってみる
準備が整ったのでAPSchedulerを使って定期実行する処理を追加します。
内容としては30秒ごとに「hi」とprintし続ける処理となり以下のコードになります。
from fastapi import FastAPI
# スケジューラで定期実行を行うのでBackgroundSchedulerをインポートする
from apscheduler.schedulers.background import BackgroundScheduler
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
# 定期実行したい処理を行う関数を作成する
def say_hi():
print("Hi")
# FastAPIを実行した際に起動するようにデコレータを指定する
@app.on_event("startup")
def skd_process():
# スケジューラのインスタンスを作成する
scheduler = BackgroundScheduler()
# スケジューラーにジョブを追加する
scheduler.add_job(say_hi, "interval", second=30)
# スケジューラを開始する
scheduler.start()特に難解な点は無いと思いますがジョブ追加の部分を少し細くしたいと思います。
ジョブを追加するのはadd_job関数にて行います。1番目のパラメータに定期実行したい処理を記述した関数名を指定します。2番目にはトリガーとしてintervalを指定しています。これはその名の通りある一定間隔にてジョブを実行するトリガーになります。他にもdateやcronなどを指定できます。最後にintervalを指定したのでどの間隔で実行するのかをsecondで指定しています。これはトリガーがdateなら具体的な日付を指定したり、cronならcron形式で*/10の様な形式で指定することも可能です。
複雑なスケジュールを組みたい場合は詳しくは公式のドキュメントをご確認ください。
まとめ
PythonのライブラリであるAPSchedulerを使ってFastAPIで定期実行を行ってみましたが特に難解な点もなく簡単に扱えました。今回は一定間隔で実行すると行った単純なスケジュール実行でしたがより複雑なスケジュールでもAPSchedulerでしたら簡単に記述できるのでおすすめいたします。
最後まで読んで頂いてありがとうございました。
2
