浜松のWEBシステム開発・スマートフォンアプリ開発・RTK-GNSS関連の開発はお任せください
株式会社シーポイントラボ
TEL:053-543-9889
営業時間:9:00~18:00(月〜金)
住所:静岡県浜松市中区富塚町1933-1 佐鳴湖パークタウンサウス2F

【Python】schedule ライブラリで Python だけで関数を定期的に自動実行する

schedule · PyPI
dbader/schedule: Python job scheduling for humans.
schedule — schedule 0.4.0 documentation
 schedule は簡易に関数を自動実行できるようになるライブラリです。Python 組み込みの sched を元に手製する必要がない時などで特に便利です。
sched — イベントスケジューラ — Python 3.8.5 ドキュメント
 具体的なコード例は次です。ファイルシステム中の特定領域を監視して違いがあったら画面にプリントするスクリプトです。

import os

import schedule
import time


def find_all_files(directory):
    """
    渡されたディレクトリパス以下のファイルとディレクトリのフルパスを list にして返す関数
    :see https://qiita.com/suin/items/cdef17e447ceeff6e79d Pythonで再帰的にファイル・ディレクトリを探して出力する - Qiita
    """
    paths = []
    for root, dirs, files in os.walk(directory):
        paths.append(root)
        for file in files:
            paths.append(os.path.join(root, file))
    return paths


def job():
    """
    あるディレクトリパス以下のファイルとディレクトリに変化が起きたらそれをプリントする関数
    """
    # 現在のあるディレクトリ以下がどうなっているか取得
    file_paths = find_all_files('d:\\xampp\\php\\CompatInfo')
    # 前回のあるディレクトリ以下がどうなっているかのログを取得
    if os.path.exists('file_paths.log'):
        with open('file_paths.log', 'r') as f:
            # readlines で読み取った時、改行コードも含まれるので strip で削除
            old_file_paths = list(map(lambda x: x.strip(), f.readlines()))
    else:
        # もしログが無ければ今回の分を記録して終了
        with open('file_paths.log', 'w') as f:
            f.writelines('\n'.join(file_paths))
        return
    # 配列の差分を取得
    # 今あるパスの中で古いパスにないモノを抜き出し
    plus = [i for i in file_paths if i not in old_file_paths]
    # 古いパスの中で今あるパスにないモノを抜き出し
    minus = [i for i in old_file_paths if i not in file_paths]

    # 実行した時に分かりやすくする様に print
    # 差分なしの場合は黙っていて、ありの時に Slack やデスクトップ通知に投げた方が便利
    if len(plus) == 0 and len(minus) == 0:
        print('差分なし')
    else:
        print('増えたパス\n' + '\n'.join(plus))
        print('減ったパス\n' + '\n'.join(minus))

    # 今回のジョブで読み取ったパスを記録して終了
    with open('file_paths.log', 'w') as f:
        f.writelines('\n'.join(file_paths))


def main():
    # 実行する関数を定義
    # schedule.毎 引数.単位.do(実行関数) な感じ
    schedule.every(5).seconds.do(job)

    while True:
        # 定義した schedule を実行しつづけます。
        schedule.run_pending()
        # とりあえず 1 秒にしてありますが、実行間隔次第ではもっと長くしても良さそうです
        time.sleep(1)


if __name__ == '__main__':
    main()

 schedule についての処理は main 関数に閉じ込められています。schedule を使うには実行関数と実行間隔を次の様に定義して

# 毎 5 秒 job 関数を実行
schedule.every(5).seconds.do(job)
# 毎 5 分 job2 関数を引数付きで実行
schedule.every(5).minutes.do(job2, arg1, arg2, key=keyarg)
# 毎水曜日13時15分に job 関数を実行。タイムゾーン指定はなさそう
# https://schedule.readthedocs.io/en/stable/faq.html?highlight=timezone#does-schedule-support-timezones
schedule.every().wednesday.at("13:15").do(job)

 次の様に while True 中で実行要求し続けるのみです

while True:
    # 定義した schedule を実行しつづけます。
    schedule.run_pending()
    # とりあえず 1 秒にしてありますが、実行間隔次第ではもっと長くしても良さそうです
    time.sleep(1)

 定期実行が必要な状況で、cron とつなげるのことができなかったり、細やかな制御をしたいがコーディングにそこまで手間をかけたくない、といった時に schedule は特に役に立ちます。

  • この記事いいね! (0)