Python Django 定期実行 APScheduler

今回は、PythonのWebフレームワークDjangoで、

定期実行の処理を実装できるAPSchedulerのメモです。

環境

• APScheduler：3.10.1
• Python version：3.10.13
• Django version：5.0
• PyCharm 2022.2.2 (Professional Edition)
  Runtime version: 17.0.4+7-b469.53 x86_64

APScheduler インストール

 pip install apscheduler

開発環境だけにインストールしたい場合、IDEの機能などでもインストールできます。

実装サンプル

構成

通常のDjangoプロジェクトのapps.pyに、定期実行を行うための修正を行います。

また、今回はap_scheduler.pyというファイルを作成して、定期実行させたい処理を実装します。

定期実行処理 実装

定期実行させてたい内容と、定期実行させる頻度などを実装します。

ap_scheduler.py

from apscheduler.schedulers.background import BackgroundScheduler


def ap_scheduler(): # 任意の関数名
    # 定期実行対象の処理を実装
    print('10秒毎に定期実行---------------------------------')


def start():
    scheduler = BackgroundScheduler()
    scheduler.add_job(ap_scheduler, 'interval', hours=0, seconds=10)
    scheduler.start()

BackgroundScheduler()について

BackgroundScheduler()でスケジューラを作成しています。
ガイドにあります通り、Django以外のフレームワークの場合、

BackgroundScheduler()の代わりに、TornadoScheduler()、TwistedScheduler()など使い分けを行います。

以下は、公式ガイドの引用ですが、BackgroundScheduler()の引数に以下を渡すことにより、
スケジューラの設定を行えます。

• 第一引数 jobstores(ジョブ保持方法)
  MemoryJobStore(メモリにジョブ保持)、SQLAlchemyJobStore(DBにジョブ保持)
• 第二引数 executors()
  ProcessPoolExecutor(ワーカープロセスの数)、ThreadPoolExecutor(起動プロセス数)
• 第三引数 job_defaults()
  max_instances(ジョブ最大稼働数)
• タイムゾーン
  utcなど

from pytz import utc from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor

jobstores = { 
    'mongo': MongoDBJobStore(), 
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite') 
}
executors = { 
    'default': ThreadPoolExecutor(20), 
    'processpool': ProcessPoolExecutor(5) 
}
job_defaults = { 
    'coalesce': False, 
    'max_instances': 3 
}

scheduler = BackgroundScheduler(
    jobstores=jobstores, 
    executors=executors, 
    job_defaults=job_defaults, 
    timezone=utc
)

add_job()について

ジョブの登録は、add_job()で行います。
https://apscheduler.readthedocs.io/en/3.x/userguide.html#adding-jobs

• 第一引数
  定期実行させたい処理を関数などで定義しておいて指定します。
• 第二引数
  定期実行ジョブのトリガーを指定します。
  interval、cron、dateなどがあります。
• 第三引数以降
  トリガーに応じて定期実行の方法などを指定します。以下は例となります。
  • interval
    sched.add_job(job_function, ‘interval’, hours=2, start_date=’2010-10-10 09:30:00′, end_date=’2014-06-15 11:00:00′)
  • cron
    sched.add_job(job_function, ‘cron’, day_of_week=’mon-fri’, hour=5, minute=30, end_date=’2014-05-30′)
  • date
    sched.add_job(job_function, ‘date’, run_date=date(2009, 11, 6), args=[‘text’])

起動などについて

起動は、start()で行います。

今回は、小さなサンプルなので利用していませんが、

一時停止pause()、再開resume()、終了shutdown()なども可能です。

apps.py 修正

Djangoのサービスが起動してから、定期実行の処理も開始させるために、

apps.pyに処理を追加します。

apps.py中のAppConfigクラスに(なければ作成)、定期実行の開始処理を追加します。

from django.apps import AppConfig


class AppConfig(AppConfig):
    name = 'djangoProject1'
    # ap_scheduler起動処理を追加
    def ready(self):
        from .ap_scheduler import start
        start()

実行確認

Djangoのサービスを起動して、定期実行されることを確認します。

定期実行が2回実行されてしまう

定期実行時に、定期実行させたい処理が必ず2回実行されてしまう現象を確認しました。

※詳細な原因を追えていませんが、Djangoのrunserver時にプロセスが2つ立ち上がる事が影響していそうです。

こちらを防ぎたい場合、以下のように –noreloadを指定すると改善されます。

manage.py runserver 0.0.0.0:8000 --noreload

PyCharmであれば、Run/Debugの設定で、Parametersに指定を加えます。

今回のメモは以上となります。

Djangoでなくても素のPythonでもAPSchedulerは利用できますし、

他のフレームワークでも対応しているものが多いです。

Pythonで定期実行させたい場合、便利です。