【Python】scheduleモジュール：定期処理の動作まとめ

はじめに

一定の周期で処理を実行する方法を調べていると、いくつか候補がでてきます。
しかし、詳細な動作を解説したページが見当たらなかった、という方も多いのではないでしょうか。

今回は、Python の schedule モジュールを使った場合の動作について、
サンプルを交えながら詳細に解説します。

結論は以下になります。

ポイント

schdule モジュールを使っても期待する定期処理は実行できない
定周期の精度は登録する Job の処理時間に左右される
定周期の精度は低い

期待する定周期処理

この記事では、以下の動作を期待する「定周期処理」として解説しています。

例えば、5秒間隔（t=5秒）に1回実行したい処理がある場合、Job が完了する時間に左右されることなく、
定期的に Job が実行されることを期待値として考えます。

schdule の基本的な使い方

schdule のインストール

 command
$ pip install schdule

使い方

多くのWebページで紹介されている通り、基本的な使い方はシンプルです。
直感的に使用することができると思います。

import schedule
import time

# 定周期処理(Job)
def job():
    print("job実行")

# 1秒毎に job 実行
schedule.every(1).seconds.do(job)

# 1分毎に job 実行
schedule.every(1).minutes.do(job)

# 1時間毎に job 実行
schedule.every(1).hours.do(job)

# 毎日AM11:00に job 実行
schedule.every().day.at("11:00").do(job)

# 毎週日曜日に job 実行
schedule.every().sunday.do(job)

# 毎週水曜日13:15に job 実行
schedule.every().wednesday.at("13:15").do(job)

# スケジューラーのメインループ
while True:
    schedule.run_pending()
    time.sleep(1)

import schedule

import time

# 定周期処理(Job)

def job():

print("job実行")

# 1秒毎に job 実行

schedule.every(1).seconds.do(job)

# 1分毎に job 実行

schedule.every(1).minutes.do(job)

# 1時間毎に job 実行

schedule.every(1).hours.do(job)

# 毎日AM11:00に job 実行

schedule.every().day.at("11:00").do(job)

# 毎週日曜日に job 実行

schedule.every().sunday.do(job)

# 毎週水曜日13:15に job 実行

schedule.every().wednesday.at("13:15").do(job)

# スケジューラーのメインループ

while True:

schedule.run_pending()

time.sleep(1)

注意ポイント

スクリプトのファイル名を「schedule.py」にしてしまうと、実行時にエラーとなります。
ファイル名は「schedule.py」以外にしましょう。

動作検証

schedule に登録した間隔より Job の処理時間は短いものの、
print だけではなく、Job の処理に時間がかかる場合の動きを見てみます。

import schedule
import time
import datetime

def dbglog(msg):
    now = datetime.datetime.now()
    print(now, msg)

def job():
    dbglog('Begin Action')
    time.sleep(3)
    dbglog('Finish Action')

# 5秒ごと
schedule.every(5).seconds.do(job)

while True:
    dbglog("Before run_pending")
    schedule.run_pending()
    dbglog("After run_pending")
    time.sleep(1)

import schedule

import time

import datetime

def dbglog(msg):

now = datetime.datetime.now()

print(now, msg)

def job():

dbglog('Begin Action')

time.sleep(3)

dbglog('Finish Action')

# 5秒ごと

schedule.every(5).seconds.do(job)

while True:

dbglog("Before run_pending")

schedule.run_pending()

dbglog("After run_pending")

time.sleep(1)

 実行結果（見やすいように加工）
00:00:00 Before run_pending
(省略)
00:00:05 Before run_pending
00:00:05 Begin Action    ★1回目
00:00:08 Finish Action
00:00:08 After run_pending
00:00:09 Before run_pending
00:00:09 After run_pending
(省略)
00:00:13 Before run_pending
00:00:13 Begin Action    ★2回目
00:00:16 Finish Action
00:00:16 After run_pending

1回目に Job が実行されるタイミングは、最初に run_pending を呼び出してから 5 秒後です。
これは期待した動きになっていることが分かります。

run_pending から戻ってくるタイミングは、Job の実行が完了した後になっています。
同一スレッドで実行しているため、ココまでは期待通りです。

しかし、2回目に Job が実行されるタイミングを見てみると、初回から 13秒後になっています。
期待値は10秒後のため、Job の処理時間に影響されて3秒の遅延が発生しています。

検証からわかること

ポイント

Job の実行に要する時間だけ、定周期から遅延して実行される
run_pending からは、Job の実行後に返ってくる

schdule の実装確認

動作検証から schdule の動きは想像できるのですが、実装も確認しておきましょう。
schdule のモジュールは、PyPi - schedule からダウンロードできます。

run_pending から戻るタイミング

def run_pending(self) -> None:
    runnable_jobs = (job for job in self.jobs if job.should_run)
    for job in sorted(runnable_jobs):
        self._run_job(job)

def run_pending(self) -> None:

runnable_jobs = (job for job in self.jobs if job.should_run)

for job in sorted(runnable_jobs):

self._run_job(job)

schdule のモジュールで run_pending は上記のように実装されています。
実行すべき Job を抽出し、順番に実行して run_pending の処理を完了する作りになっています。

そのため、動作検証での結果の通り、Job の実行が完了したのち run_pending から返ってきます。

Job の実行タイミングを決めるタイミング

@property
def should_run(self) -> bool:
    return datetime.datetime.now() >= self.next_run

def _run_job(self, job: "Job") -> None:
    job.run()

def run(self):
    ret = self.job_func()
    self.last_run = datetime.datetime.now()
    self._schedule_next_run()
    return ret

@property

def should_run(self) -> bool:

return datetime.datetime.now() >= self.next_run

def _run_job(self, job: "Job") -> None:

job.run()

def run(self):

ret = self.job_func()

self.last_run = datetime.datetime.now()

self._schedule_next_run()

return ret

run_pending の中で実行すべき Job を抽出する際、should_run をチェックしています。
should_run は、next_run が現在時刻を経過していたら True になる値です。

次回の実行時刻を示す next_run は、job_func() で Job が実行された後の時刻 last_run に、
最初に設定した「5秒毎」などの周期が加算されて計算されます。

つまり、Job の実行が完了した時刻に「周期」時刻が加算され、
次回 run_pending が呼び出された際に実行される流れとなります。

おわりに

いかがでしたでしょうか。
schdule や every という単語だけを見て使用すると、痛い目にあいそうですね。

Job の処理に 3秒以上かかる例で解説しましたが、schdule 内部の処理や print の処理だけでも
定周期で何回も実行して蓄積されれば秒単位で「実行周期」がずれることになります。

どのように実装すれば期待通りの定周期処理が行えるのかは、
次回の投稿で紹介したいと思います。