Pythonのscheduleで定期実行を実装する方法とサンプル

# 仮想環境の利用をおすすめします
python -m venv .venv
# Windows
.venv\Scripts\activate
# macOS/Linux
source .venv/bin/activate

# scheduleのインストール
pip install schedule

# 参考: バージョン確認
pip show schedule

# file: minimal_schedule.py
import schedule
import time
from datetime import datetime

def hello():
    """定期的に呼び出される関数"""
    print(f"[{datetime.now().strftime('%H:%M:%S')}] Hello, schedule!")

# 5秒ごとにhelloを実行
schedule.every(5).seconds.do(hello)

# ループして実行時刻になったジョブを動かす
# Ctrl+Cで停止できます
try:
    while True:
        schedule.run_pending()  # そろそろ実行すべきジョブがあれば実行
        time.sleep(1)           # CPUを占有しないように少し待つ
except KeyboardInterrupt:
    print("Stopped.")

import schedule
import time

def job_minute():
    print("毎分のジョブを実行")

def job_hour():
    print("毎時のジョブを実行")

# 毎分
schedule.every().minute.do(job_minute)

# 5分ごと
schedule.every(5).minutes.do(lambda: print("5分おきのジョブ"))

# 毎時
schedule.every().hour.do(job_hour)

# 各ループ
while True:
    schedule.run_pending()
    time.sleep(1)

# 各時間の30分に実行 (例: 10:30, 11:30, 12:30, ...)
schedule.every().hour.at(":30").do(lambda: print("各時間の30分に実行"))

import schedule
from datetime import datetime

def backup():
    print(f"{datetime.now()} バックアップを実行中...")

schedule.every().day.at("09:00").do(backup)

import schedule

def weekday_task():
    print("平日だけのタスク")

# 曜日を個別に登録 (平日のみ)
schedule.every().monday.at("10:00").do(weekday_task)
schedule.every().tuesday.at("10:00").do(weekday_task)
schedule.every().wednesday.at("10:00").do(weekday_task)
schedule.every().thursday.at("10:00").do(weekday_task)
schedule.every().friday.at("10:00").do(weekday_task)

# 週次 (毎週火曜の13:15)
schedule.every().tuesday.at("13:15").do(lambda: print("週次タスク"))
# 2週間ごと (隔週金曜の18:00) も可能
schedule.every(2).weeks.do(lambda: print("隔週タスク")).at("18:00")

import schedule
import time

def job_a():
    print("Aを実行")

def job_b():
    print("Bを実行")

# タグを付けて登録
schedule.every(2).minutes.do(job_a).tag("group1", "fast")
schedule.every().hour.do(job_b).tag("group1", "slow")

# タグで停止・削除も可能
def stop_group1():
    print("group1をクリア")
    schedule.clear("group1")

# 実験用に10分後に削除を仕込む
schedule.every(10).minutes.do(stop_group1).tag("admin")

while True:
    schedule.run_pending()
    time.sleep(1)

import schedule
from datetime import datetime

def greet(name, punctuation="!"):
    print(f"{datetime.now():%H:%M:%S} Hello, {name}{punctuation}")

# 3分おきに引数付きで実行
schedule.every(3).minutes.do(greet, "Alice", punctuation="!!!")

import schedule
import time

def loop_forever():
    try:
        while True:
            schedule.run_pending()
            time.sleep(1)
    except KeyboardInterrupt:
        print("Stopped.")

loop_forever()

import schedule
import time
import math

def efficient_loop():
    try:
        while True:
            schedule.run_pending()
            idle = schedule.idle_seconds()  # 次のジョブまでの秒数 or None
            if idle is None:
                time.sleep(1)
            else:
                # 0以下の場合はすぐにrun_pendingに戻る
                time.sleep(max(0.0, min(idle, 60)))  # 長くても60秒まで
    except KeyboardInterrupt:
        print("Stopped.")

efficient_loop()

import schedule
import time
from random import random

def with_retry(task, *, retries=3, base_delay=1.0, factor=2.0):
    """指数バックオフで簡易リトライを行うラッパー"""
    def wrapper(*args, **kwargs):
        delay = base_delay
        for attempt in range(1, retries + 1):
            try:
                return task(*args, **kwargs)
            except Exception as e:
                print(f"[{task.__name__}] エラー: {e} (試行{attempt}/{retries})")
                if attempt == retries:
                    print(f"[{task.__name__}] 失敗を記録して中断")
                    break
                time.sleep(delay)
                delay *= factor
    return wrapper

@with_retry
def flaky_job():
    # たまに失敗する処理の例
    if random() < 0.5:
        raise RuntimeError("一時的な失敗")
    print("成功!")

schedule.every(30).seconds.do(flaky_job)

try:
    while True:
        schedule.run_pending()
        time.sleep(1)
except KeyboardInterrupt:
    print("Stopped.")

import schedule

counter = 0
def limited_job():
    global counter
    counter += 1
    print(f"実行{counter}回目")
    if counter >= 3:
        print("このジョブは終了します")
        return schedule.CancelJob

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

import schedule
import threading
import time

lock = threading.Lock()

def long_running_task():
    # 今動作中ならスキップ
    if not lock.acquire(blocking=False):
        print("前回の処理がまだ終わっていないためスキップしました")
        return
    try:
        print("長時間処理を開始...")
        time.sleep(10)  # 実際の重い処理を想定
        print("長時間処理が完了しました")
    finally:
        lock.release()

# 1分おきに起動要求が来ても、重複は起こさない
schedule.every().minute.do(long_running_task)

while True:
    schedule.run_pending()
    time.sleep(1)

from datetime import datetime
from zoneinfo import ZoneInfo
import schedule
import time

last_run_date = None  # 同じ日での二重実行防止

def notify_tokyo_1730():
    global last_run_date
    now_tokyo = datetime.now(ZoneInfo("Asia/Tokyo"))
    # 17:30ちょうどの分で、未実行なら実行
    if now_tokyo.hour == 17 and now_tokyo.minute == 30:
        if last_run_date != now_tokyo.date():
            print(f"Tokyo 17:30 通知を送信: {now_tokyo.isoformat()}")
            last_run_date = now_tokyo.date()
        else:
            # 同じ分内で複数回呼ばれないようガード
            pass

# 毎分チェック
schedule.every().minute.do(notify_tokyo_1730)

while True:
    schedule.run_pending()
    time.sleep(1)

# file: heartbeat_every_minute.py
import schedule
import time
import logging
from datetime import datetime

# ロガー設定
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s"
)

def heartbeat():
    """アプリが生きていることを示すハートビート"""
    logging.info("heartbeat")

# 毎分実行
schedule.every().minute.do(heartbeat)

def main():
    logging.info("start scheduler")
    try:
        while True:
            schedule.run_pending()
            time.sleep(1)
    except KeyboardInterrupt:
        logging.info("stopped by user")

if __name__ == "__main__":
    main()

# file: daily_backup_0900.py
import schedule
import time
import shutil
from datetime import datetime
from pathlib import Path

SRC = Path("data")                       # バックアップ元
DST_BASE = Path("backup")                # バックアップ先のルート
RETRIES = 3

def dated_backup_dir(base: Path, now: datetime) -> Path:
    """日付ベースのバックアップディレクトリ"""
    return base / now.strftime("%Y%m%d_%H%M%S")

def backup():
    """ディレクトリを丸ごとコピーするバックアップ"""
    now = datetime.now()
    dst = dated_backup_dir(DST_BASE, now)
    for attempt in range(1, RETRIES + 1):
        try:
            dst.parent.mkdir(parents=True, exist_ok=True)
            print(f"[{now}] backup start -> {dst}")
            shutil.copytree(SRC, dst)  # 存在する場合は例外
            print(f"[{datetime.now()}] backup done")
            return
        except FileExistsError:
            # 同一時刻での重複を避けるため1秒待って再生成
            time.sleep(1)
            dst = dated_backup_dir(DST_BASE, datetime.now())
        except Exception as e:
            print(f"backup error: {e} (attempt {attempt}/{RETRIES})")
            time.sleep(2 ** attempt)
    print("backup failed after retries")

# 毎日9:00にバックアップ
schedule.every().day.at("09:00").do(backup)

if __name__ == "__main__":
    print("scheduler started")
    try:
        while True:
            schedule.run_pending()
            time.sleep(1)
    except KeyboardInterrupt:
        print("stopped")

# file: notify_weekdays_1730.py
import schedule
from datetime import datetime
import time

def send_notification():
    now = datetime.now()
    print(f"{now:%Y-%m-%d %H:%M} 平日17:30の通知を送信しました")

# 月〜金の17:30に登録
for day in (schedule.every().monday,
            schedule.every().tuesday,
            schedule.every().wednesday,
            schedule.every().thursday,
            schedule.every().friday):
    day.at("17:30").do(send_notification)

if __name__ == "__main__":
    print("notification scheduler started")
    try:
        while True:
            schedule.run_pending()
            time.sleep(1)
    except KeyboardInterrupt:
        print("stopped")

# file: manage_multiple_jobs.py
import schedule
import time
import signal
from datetime import datetime
from typing import Callable, Any

STOP = False  # 終了フラグ

def setup_signal_handlers():
    """Ctrl+CやSIGTERMで安全に停止する"""
    def handler(signum, frame):
        global STOP
        print(f"signal received: {signum}, stopping...")
        STOP = True
    signal.signal(signal.SIGINT, handler)
    # WindowsではSIGTERMが無い場合があるため存在チェック
    if hasattr(signal, "SIGTERM"):
        signal.signal(signal.SIGTERM, handler)

def add_job(func: Callable[..., Any], spec: str, *, tag: str, **kwargs):
    """
    spec: 簡易的な頻度指定
        - "min"         -> 毎分
        - "5min"        -> 5分おき
        - "hour:30"     -> 毎時30分
        - "daily@09:00" -> 毎日09:00
    kwargs: funcに渡すキーワード引数
    """
    if spec == "min":
        return schedule.every().minute.do(func, **kwargs).tag(tag)
    if spec.endswith("min") and spec[:-3].isdigit():
        n = int(spec[:-3])
        return schedule.every(n).minutes.do(func, **kwargs).tag(tag)
    if spec.startswith("hour:"):
        mm = spec.split(":", 1)[1]
        return schedule.every().hour.at(f":{mm}").do(func, **kwargs).tag(tag)
    if spec.startswith("daily@"):
        t = spec.split("@", 1)[1]
        return schedule.every().day.at(t).do(func, **kwargs).tag(tag)
    raise ValueError(f"unsupported spec: {spec}")

def job_log(message: str):
    print(f"[{datetime.now():%H:%M:%S}] {message}")

def job_sum(a: int, b: int):
    print(f"{a} + {b} = {a + b}")

def main():
    setup_signal_handlers()

    # ログを毎分
    add_job(job_log, "min", tag="log", message="heartbeat")

    # 毎時30分にお知らせ
    add_job(job_log, "hour:30", tag="news", message="30分の定期メッセージ")

    # 毎日09:00に計算実行(引数付き)
    add_job(job_sum, "daily@09:00", tag="calc", a=40, b=2)

    print("jobs loaded:")
    for j in schedule.get_jobs():
        print(" -", j)

    # 実行ループ
    while not STOP:
        schedule.run_pending()
        time.sleep(1)

    # 終了時に特定タグのみクリアする例
    schedule.clear("news")
    print("scheduler stopped")

if __name__ == "__main__":
    main()