Python scheduleライブラリ徹底解説！🐍 ジョブスケジューリングを簡単実装

はじめに：scheduleライブラリとは？ 🤔

Pythonで「毎週月曜日の朝9時に特定の処理を実行したい」「10分ごとにAPIからデータを取得したい」といった定期的なタスク（ジョブ）を実行したい場面はよくあります。そんな時に役立つのが schedule ライブラリです。

schedule は、人間にとって非常に直感的で読みやすい構文でジョブスケジューリングを定義できることを目指して設計されています。複雑な設定ファイルやデーモンの知識がなくても、Pythonスクリプト内で簡単に定期実行処理を組み込めるのが大きな魅力です 😊。

このライブラリは、特にシンプルなスケジューリング要件を持つ小〜中規模のアプリケーションや、バッチ処理、定期的なデータ収集・通知などのタスクに適しています。Linuxの cron やWindowsのタスクスケジューラのようなOSレベルのスケジューラとは異なり、Pythonプロセス内で動作するため、スクリプトが実行されている間だけジョブがスケジュールされます。

この記事では、schedule ライブラリのインストール方法から基本的な使い方、少し応用的なテクニック、そして利用する上での注意点や代替手段まで、幅広く詳しく解説していきます。さあ、schedule を使って、あなたのPythonスクリプトに自動化の魔法をかけましょう！✨

基本的な使い方 🕒

schedule の基本的な使い方は、以下のステップで行います。

1. ライブラリをインポートする (schedule と time)。
2. 定期実行したい処理を関数として定義する。
3. schedule.every() メソッドチェーンを使って、いつ、どの関数を実行するかを定義する。
4. 無限ループ内で schedule.run_pending() を呼び出し、スケジュールされたジョブを確認・実行する。
5. ループ内で time.sleep() を使って、CPUを過剰に消費しないように待機する。

1. インポート

まず、必要なライブラリをインポートします。

import schedule
import time
import datetime

print(f"スクリプト開始: {datetime.datetime.now()}")

2. ジョブ関数の定義

定期的に実行したい処理を関数として定義します。引数を取らないシンプルな関数を例にします。

def job():
    """定期実行したい処理を記述する関数"""
    current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f"ジョブ実行中！ 現在時刻: {current_time}")

3. スケジューリング

schedule.every() を起点として、メソッドチェーンで実行タイミングを指定します。非常に直感的です。

# --- 様々なスケジューリング例 ---

# 10秒ごとに job 関数を実行
schedule.every(10).seconds.do(job)

# 10分ごとに job 関数を実行
# schedule.every(10).minutes.do(job)

# 毎時間ごとに job 関数を実行 (例: 1:00, 2:00, ...)
# schedule.every().hour.do(job)

# 毎日 10:30 に job 関数を実行
# schedule.every().day.at("10:30").do(job)

# 毎週月曜日に job 関数を実行
# schedule.every().monday.do(job)

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

# 毎分17秒のタイミングで job 関数を実行 (例: 10:00:17, 10:01:17, ...)
# schedule.every().minute.at(":17").do(job)

# 毎時30分に job 関数を実行 (例: 09:30:00, 10:30:00, ...)
# schedule.every().hour.at(":30").do(job)

# 5〜10分(ランダム)ごとに job 関数を実行
# schedule.every(5).to(10).minutes.do(job)

print("スケジュールの設定が完了しました。")
print(f"現在スケジュールされているジョブ: {schedule.get_jobs()}")

このように、every() の後に単位 (.seconds, .minutes, .hours, .days, .weeks) や曜日 (.monday, .tuesday, etc.) を指定し、さらに .at("HH:MM") や .at(":SS") で特定の時刻や秒を指定できます。.do(関数名) で実行する関数を結びつけます。

4 & 5. メインループと実行

最後に、スケジュールされたジョブを実行するためのループを作成します。

print("スケジューリングループを開始します...")
try:
    while True:
        # 現在時刻で実行されるべきジョブがあるか確認し、あれば実行
        schedule.run_pending()
        # 1秒待機して、CPUの負荷を下げる
        time.sleep(1)
except KeyboardInterrupt:
    print("\nプログラムがユーザーによって中断されました。")
    # スケジュールをクリア
    schedule.clear()
    print("スケジュールをクリアしました。")
finally:
    print(f"スクリプト終了: {datetime.datetime.now()}")

この while True: ループがスクリプトの心臓部です。schedule.run_pending() は、現在時刻を見て、実行すべきスケジュールされたジョブがあれば実行します。なければ何もせず、すぐに処理が戻ります。time.sleep(1) は、1秒ごとに run_pending() をチェックするようにしています。この値を小さくすればより高頻度にチェックしますが、CPU使用率は上がります。逆に大きくすると、ジョブの実行タイミングのずれが大きくなる可能性がありますが、CPU負荷は下がります。通常は1秒程度で十分な場合が多いでしょう。

上記のコードを結合して実行すると、10秒ごとにコンソールに「ジョブ実行中！ 現在時刻: …」と表示されるはずです。Ctrl+C などでスクリプトを中断すると、KeyboardInterrupt 例外が発生し、ループが終了します。

実践的な例 💡

これまでに学んだ機能を使って、いくつかの実践的な例を見てみましょう。

例1: 毎日特定の時間にメールレポートを送信

ここでは、メール送信自体の実装は省略し、スケジューリング部分に焦点を当てます。

import schedule
import time
import datetime
# import smtplib # 実際のメール送信に必要
# from email.message import EmailMessage # 実際のメール送信に必要

def send_daily_report():
    """日次レポートを生成してメールで送信する関数（ダミー）"""
    print(f"[{datetime.datetime.now()}] 日次レポートを生成・送信しています...")
    # ここでレポート生成ロジック (例: DBから集計)
    report_content = f"本日の売上サマリーです。\n（これは {datetime.datetime.now()} に生成されました）"
    # ここでメール送信ロジック (smtplibなどを使用)
    # msg = EmailMessage()
    # msg.set_content(report_content)
    # msg['Subject'] = f'{datetime.date.today()} の日次レポート'
    # msg['From'] = 'report@example.com'
    # msg['To'] = 'manager@example.com'
    # smtp_server = 'smtp.example.com'
    # smtp_port = 587
    # username = 'user'
    # password = 'password'
    # with smtplib.SMTP(smtp_server, smtp_port) as server:
    #     server.starttls()
    #     server.login(username, password)
    #     server.send_message(msg)
    print("レポート送信完了。")

# 毎日午前8時30分にレポート送信ジョブを実行
schedule.every().day.at("08:30").do(send_daily_report).tag("daily-report")

print("日次レポート送信ジョブをスケジュールしました。")
print(f"次の実行予定: {schedule.next_run}")

try:
    while True:
        schedule.run_pending()
        time.sleep(60) # 1分ごとにチェック (メール送信なので秒単位の精度は不要)
except KeyboardInterrupt:
    print("\n終了します。")
finally:
    schedule.clear()

この例では、every().day.at("08:30") を使って、毎日午前8時30分に send_daily_report 関数を実行するようにスケジュールしています。time.sleep() の間隔は、ジョブの実行頻度に合わせて調整します。1日1回のジョブであれば、60秒やそれ以上の間隔でも問題ないでしょう。

例2: 1時間ごとに一時ファイルをクリーンアップ

import schedule
import time
import datetime
import os
import glob

TEMP_DIR = "./temp_files" # 一時ファイルが保存されるディレクトリ (例)

def cleanup_temp_files():
    """TEMP_DIR 内の古い一時ファイルを削除する関数"""
    print(f"[{datetime.datetime.now()}] 一時ファイルのクリーンアップを開始します...")
    now = time.time()
    # 古いファイルの閾値（例：1日以上前）
    threshold_age_seconds = 24 * 60 * 60

    if not os.path.exists(TEMP_DIR):
        print(f"ディレクトリが存在しません: {TEMP_DIR}")
        return

    try:
        for filename in glob.glob(os.path.join(TEMP_DIR, '*')):
            if os.path.isfile(filename):
                file_mtime = os.path.getmtime(filename)
                if now - file_mtime > threshold_age_seconds:
                    print(f"  古いファイルを削除します: {filename} (最終更新: {datetime.datetime.fromtimestamp(file_mtime)})")
                    # os.remove(filename) # 実際に削除する場合はコメントアウトを外す
                else:
                     print(f"  比較的新しいファイルのため保持: {filename}")
    except Exception as e:
        print(f"クリーンアップ中にエラーが発生しました: {e}")
    finally:
        print("クリーンアップ処理を完了しました。")

# 毎時0分にクリーンアップジョブを実行
schedule.every().hour.at(":00").do(cleanup_temp_files).tag("cleanup")

# スクリプト開始時に一度実行しておく場合
# print("初回クリーンアップを実行します...")
# cleanup_temp_files()
# もしくは schedule.run_all() を使う
# schedule.run_all()

print("一時ファイルクリーンアップジョブをスケジュールしました。")
print(f"次の実行予定: {schedule.next_run}")

# テスト用に一時ファイルディレクトリとファイルを作成
if not os.path.exists(TEMP_DIR):
    os.makedirs(TEMP_DIR)
# 古いファイルと新しいファイルの例を作成
with open(os.path.join(TEMP_DIR, "old_file.tmp"), "w") as f: f.write("old")
# 過去のタイムスタンプを設定 (2日前)
two_days_ago = time.time() - 2 * 24 * 60 * 60
os.utime(os.path.join(TEMP_DIR, "old_file.tmp"), (two_days_ago, two_days_ago))
with open(os.path.join(TEMP_DIR, "new_file.tmp"), "w") as f: f.write("new")


try:
    while True:
        schedule.run_pending()
        time.sleep(1) # 1時間ごとのジョブだが、より正確な時刻実行のため sleep は短めに設定
except KeyboardInterrupt:
    print("\n終了します。")
finally:
    # テスト用ファイルを削除
    # import shutil
    # if os.path.exists(TEMP_DIR): shutil.rmtree(TEMP_DIR)
    schedule.clear()

この例では、every().hour.at(":00") を使用して、毎時ちょうど（例: 9:00:00, 10:00:00）に一時ファイル削除の関数を実行します。at(":00") の代わりに every().hour だけだと、スクリプトを開始した時刻基準の毎時（例: 9:15に開始したら、10:15, 11:15…）になります。目的に応じて使い分けましょう。

例3: 5分ごとにAPIからデータを取得

import schedule
import time
import datetime
import requests # データ取得に requests ライブラリを使用 (pip install requests)
import json

API_ENDPOINT = "https://api.coindesk.com/v1/bpi/currentprice.json" # 例: Bitcoin価格API

def fetch_bitcoin_price():
    """Bitcoinの現在価格を取得して表示する関数"""
    print(f"[{datetime.datetime.now()}] Bitcoin価格を取得します...")
    try:
        response = requests.get(API_ENDPOINT)
        response.raise_for_status() # エラーがあれば例外を発生させる (4xx, 5xx)
        data = response.json()
        usd_price = data['bpi']['USD']['rate']
        timestamp = data['time']['updatedISO']
        print(f"取得成功！ 現在のBitcoin価格 (USD): {usd_price} (取得時刻: {timestamp})")
        # ここで取得したデータをDBに保存したり、ファイルに書き出したりする処理を追加できる
        # save_to_database(timestamp, usd_price)
    except requests.exceptions.RequestException as e:
        print(f"APIからのデータ取得中にエラーが発生しました: {e}")
    except KeyError as e:
        print(f"受信データの形式が予期したものと異なります: {e}")
    except Exception as e:
        print(f"予期せぬエラーが発生しました: {e}")

# 5分ごとに価格取得ジョブを実行
schedule.every(5).minutes.do(fetch_bitcoin_price).tag("api-fetch")

print("Bitcoin価格取得ジョブを5分ごとに実行するようにスケジュールしました。")
print(f"次の実行予定: {schedule.next_run}")

# スクリプト開始時にも一度実行する
# fetch_bitcoin_price()
# または
# schedule.run_all()

try:
    while True:
        schedule.run_pending()
        time.sleep(1) # 1秒ごとにチェック
except KeyboardInterrupt:
    print("\n終了します。")
finally:
    schedule.clear()

every(5).minutes を使って、5分間隔でAPIからデータを取得する関数を実行します。APIリクエストには時間がかかる可能性や、ネットワークエラーが発生する可能性もあるため、try...except ブロックで適切にエラーハンドリングを行うことが重要です。

代替手段 🤔

schedule はシンプルさが魅力ですが、より複雑な要件や堅牢性が求められる場合には、他のツールやライブラリを検討する価値があります。

どのツールを選択するかは、プロジェクトの要件（シンプルさ、機能性、堅牢性、スケーラビリティ、永続化の必要性など）に応じて決定することが重要です。シンプルなユースケースであれば schedule が手軽で良い選択肢となります 👍。