今回は、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’])
- interval
起動などについて
起動は、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で定期実行させたい場合、便利です。