【Django】APSchedulerで定期実行する方法|misfire_grace_timeエラーの解決も解説

2022年7月25日

Djangoアプリで「1時間ごとにデータを取得してDBに保存する」「毎朝4時に重複チェックを実行する」など、バックグラウンドで処理を定期実行したい場面があります。

本記事ではAPSchedulerを使ったDjangoの定期実行設定を、インストールから「was missed by」エラーへの対処まで解説します。

APSchedulerをインストール

APSchedulerは軽量で使いやすいPythonのジョブスケジューラーです。

pip install apscheduler

実行したいコードを記述する

アプリフォルダ内にtasksフォルダを作成し、task.pyを作成します。

app/
├── admin.py
├── models.py
├── views.py
└── tasks/
    └── task.py

task.pyに以下のように記述します。

from apscheduler.schedulers.background import BackgroundScheduler

def periodic_execution():
    # ここに定期実行したい処理を記述する
    print("定期実行")

def start():
    scheduler = BackgroundScheduler()
    scheduler.add_job(periodic_execution, 'cron', hour=8, minute=30)  # 毎日8時30分に実行
    scheduler.start()

# --- スケジュール設定の例 ---
# 5分おきに実行
# scheduler.add_job(periodic_execution, 'interval', minutes=5)
# 1時間おきに実行
# scheduler.add_job(periodic_execution, 'interval', hours=1)
# 1日おきに実行
# scheduler.add_job(periodic_execution, 'interval', days=1)
# 毎時20分に実行
# scheduler.add_job(periodic_execution, 'cron', minute=20)
# 月曜から金曜の8時に実行
# scheduler.add_job(periodic_execution, 'cron', hour=8, day_of_week='mon-fri')
# 期間を指定して実行
# scheduler.add_job(periodic_execution, 'interval', minutes=1,
#     start_date="2024-04-01 19:00:00",
#     end_date="2024-04-01 20:00:00")

サーバー起動時に処理が定期実行されるようにする

アプリフォルダ内のapps.pyに以下のコードを追記します。Djangoサーバーの起動と同時にスケジューラーが開始されます。

from django.apps import AppConfig


class AppConfig(AppConfig):
    name = 'アプリ名'
    
    def ready(self):
        from .tasks.task import start
        start()

「〜was missed by」エラーが発生して実行失敗する場合

デフォルトでは、予定された実行時刻より1秒以上遅れた場合にジョブがスキップされます。サーバー起動直後や負荷が高い環境で発生しやすいエラーです。

解決策:misfire_grace_timeを設定して許容する遅延時間(秒)を指定します。

# 300秒(5分)以内の遅延を許容する
scheduler.add_job(periodic_execution, 'cron', hour=8, minute=30, misfire_grace_time=300)

misfire_grace_time=Noneを指定すると遅延時間を無制限に許容します。

本番環境での注意点

  • プロセス二重起動:Gunicornのワーカーが複数ある場合、スケジューラーが複数起動して処理が重複実行されることがあります。本番環境ではdjango-apschedulerパッケージやCelery Beatの利用を検討してください。
  • 開発サーバーでの二重実行対策
def ready(self):
    import os
    if os.environ.get('RUN_MAIN'):  # 開発サーバーの二重起動対策
        from .tasks.task import start
        start()

Django,python

Posted by Next-k