プログラミング原人の進化ログ

プログラミング原人の進化論

オレ プログラミング ベンキョウ スル。マナンダ コト カク。

APSchedulerの使い方(初心者向け)

APSchedulerの使い方のメモです。
APSchedulerはPythonのライブラリで、ジョブの自動実行のスケジュール管理を行なってくれるものです。この記事ではインストールから、基本的な使い方までを見てみます。

動機

  • 日本語で使い方を体系的に説明した資料が少ないので、なるべくわかりやすくまとめておく
  • 備忘録

環境

Mac OS Mojave バージョン 10.14.3

APSchedulerで何ができる?

指定したタイミングで指定した処理(ジョブ)を自動的に行ってくれます。実行するタイミングは日時、インターバルなど好きなように設定できます。

インストール

pipでインストールできます。

pip install apscheduler

使い方

仕組み

APSchedulerは次のようなコンポーネントを備えており、それぞれが別々の役割を担っています。

  • trigger:ジョブを実行するタイミングを指定
  • job store:ジョブを保管しておく
  • executor:ジョブを実行する
  • scheduler:上の3つを統括する

ジョブの実行に直接関わるのはtrigger, job store, executorですが、ユーザはschedulerを通してこれらを操作することになります。
コンポーネントはそれぞれ複数の種類が用意されていて、実行したい処理に応じて適切に選択することが求められますが、この記事ではそれらのうちほんの一部しか扱いません。具体的には、明示的に指定するのはスケジューラとトリガーにとどめ、他はデフォルトとします。また、スケジューラは最も基本的なBlockingSchedulerを使います。

詳しい機能は公式リファレンスで。

基本的な流れ

  • スケジューラのインスタンスを作成
  • 実行すべきジョブのリストにジョブを加える
  • スケジューラをスタートさせる

例を見てみましょう。3秒ごとに現在時刻を表示するプログラムです。公式からほとんどそのまま引っ張ってきました。

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
import os


def tick():
    print("Tick! The time is : %s'" % datetime.now())


if __name__ == '__main__':
    scheduler = BlockingScheduler()  # スケジューラを作る
    scheduler.add_job(tick, 'interval', seconds=3)  # ジョブを追加
    print(
        "Press Ctrl+{0} to exit.".format('Break' if os.name == 'nt' else 'C'))

    try:
        scheduler.start()  # スタート
    except (KeyboardInterrupt, SystemExit):
        pass

実行結果:

$ python scheduler_blocking.py 
Press Ctrl+C to exit.
Tick! The time is : 2019-03-31 20:33:30.974106'
Tick! The time is : 2019-03-31 20:33:33.973824'
Tick! The time is : 2019-03-31 20:33:36.973688'
Tick! The time is : 2019-03-31 20:33:39.976681'
Tick! The time is : 2019-03-31 20:33:42.973628'
Tick! The time is : 2019-03-31 20:33:45.976807'
Tick! The time is : 2019-03-31 20:33:48.972867'
^C

ジョブの追加にはadd_job()メソッドを使います。ここでジョブや、実行するタイミングを指定します。通常、ジョブの処理内容は上の例のように関数を利用して指定するようです。また、2番目の引数が'interval'となっていますが、これはトリガーの種類を表しています。トリガーのパラメータを指定しますが、ここでは3番目の引数secondで実行間隔を指定していますね。これによって一定のインターバルでジョブを実行することができます。

スケジューラをスタートさせるにはstart()メソッドを使います。

トリガーを指定する

様々な方法でジョブが発火するタイミングを指定できます。トリガーを使います。

一定間隔でジョブを実行する

add_job()メソッドの引数で、トリガーに'interval'を指定します。seconds, minutes, hours, daysなどのパラメータがあります。start_time, end_timeで開始や終了のタイミングを指定することもできます。
詳しくはこちら

時刻を指定してジョブを実行する

トリガーに'date'を指定します。実行タイミングはrun_dateで指定します。

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

scheduler = BlockingScheduler()


def my_job(text):
    print("The time is: %s" % datetime.now())
    print(text)


scheduler.add_job(
    my_job, 'date', run_date="2019-03-31 21:45:00", args=['Hello'])

scheduler.start()

実行結果:

The time is: 2019-03-31 21:45:00.001948
Hello

詳しくは こちら

cronライクにトリガーを指定する

トリガーを'cron'にします。Unix系OSで使われるcronのような形式で実行タイミングを指定します。
cronは詳しく説明すると大変なので興味のある方はこちらから。決して僕が説明できないと言うわけでは...。

まとめ

APSchedulerの基本的な使い方をまとめました。あまり深掘りはできませんでしたが、初心者の方には十分かな、ということで。詳しい使い方は適宜公式を参照してもらえればと思います。Twitterbotやなんかを作るのに役立ちそう。あとcronも勉強しておきたいです。

参考

Advanced Python Scheduler — APScheduler 3.0.2 documentation