Day 42:定期実行で「自動で動き続ける仕組み」を考える
Day 42では、業務自動化の中でもとても重要な 「定期実行」 をテーマにします。 これまで作ってきた「CSV変換ツール」「集計ツール」「PDF操作ツール」などは、 手動で python xxx.py を実行していましたが、 本当に業務を楽にするには「毎日決まった時間に勝手に動いてくれる」仕組みが欲しくなります。
ここでは、次のキーワードを軸に進めていきます。
- スケジュール処理
- バッチ処理
- 定期実行の考え方
Pythonだけで完結する「簡易的な定期実行」と、 OSの機能(cronやタスクスケジューラ)と組み合わせる「本格的な定期実行」の考え方を、 初心者向けにかみ砕いて説明していきます。
定期実行とは何かを言葉で整理する
「人が押す実行ボタン」を「時間」が押してくれるイメージ
これまでのスクリプトは、次のような流れでした。
- 人がコマンドを打つ(例:
python day39_csv_transform_tool.py) - スクリプトが1回だけ動く
- 終わったら止まる
定期実行 は、これを次のように変えるイメージです。
- 「毎日朝9時に自動で動く」
- 「1時間ごとにログを集計する」
- 「毎週月曜日にレポートを作る」
つまり、「時間をトリガーにしてスクリプトを動かす」 という考え方です。
バッチ処理とは何か
「まとめて一気に処理する」スタイル
バッチ処理 という言葉は、業務システムでよく出てきます。
- 昼間はシステムを使っている人が多いので、 夜間に「1日のデータをまとめて集計する」
- 月末に「1ヶ月分の請求書をまとめて発行する」
こうした「あるタイミングで、まとめて一気に処理する」スタイルがバッチ処理です。
Pythonで作る「CSV変換ツール」「集計ツール」などは、 まさにバッチ処理の一種と言えます。
スケジュール処理の基本的な考え方
1. 「何を」定期実行したいかを決める
まずは、定期実行したい処理を1つの関数にまとめます。
例:前日に作られたCSVを集計してレポートを作る処理。
# day42_batch_job.py
def run_daily_job():
"""毎日実行したいバッチ処理の例です。"""
print("=== 日次バッチ処理を開始します ===")
# ここに「CSV読み込み → データ加工 → 集計 → 出力」などの処理を書きます。
# 今はサンプルとしてメッセージだけにしておきます。
print("CSVを読み込みます...")
print("データを加工します...")
print("集計結果を出力します...")
print("=== 日次バッチ処理が完了しました ===")
Python2. 「いつ」「どのくらいの頻度で」実行するかを決める
- 毎日1回
- 1時間ごと
- 5分ごと
など、頻度を決めます。
3. その頻度で関数を呼び出す仕組みを作る
ここからが「スケジュール処理」の話になります。
Pythonだけで作る簡易的な定期実行
time.sleep() を使った「簡易ループ」
Python標準ライブラリの time モジュールを使うと、 「一定時間待ってから処理を実行する」ということができます。
# day42_simple_scheduler.py
import time
from datetime import datetime
def run_daily_job():
"""毎回実行したい処理(バッチ処理)の例です。"""
print("=== バッチ処理開始 ===")
print("現在時刻:", datetime.now())
print("ここでCSV集計やレポート生成などを行います。")
print("=== バッチ処理終了 ===\n")
def main():
# ここでは「60秒ごとに実行する」簡易スケジューラを作ります。
interval_seconds = 60
print("=== 簡易スケジューラを開始します ===")
print(f"{interval_seconds}秒ごとにバッチ処理を実行します。")
while True:
run_daily_job() # バッチ処理を実行
time.sleep(interval_seconds) # 指定秒数だけ待機
if __name__ == "__main__":
main()
Pythonポイント:
while True:で「永遠に繰り返す」ループを作っている。time.sleep(秒数)で「その秒数だけ待つ」。- この例では「60秒ごと」に
run_daily_job()が呼ばれます。
これは「本番用のスケジューラ」というより、 「定期実行の感覚をつかむための練習用」 として捉えると良いです。
「毎日決まった時間に実行する」考え方
現在時刻を見て「次の実行時刻まで待つ」
もう少し踏み込んで、「毎日9時に実行する」イメージを考えてみます。
- 現在時刻を取得する
- 今日の9時を表す
datetimeを作る - 「今」が9時より前なら、その差分だけ
sleepする - 「今」が9時を過ぎているなら、翌日の9時までの差分を計算して
sleepする
これをコードにすると、少し長くなりますが、考え方はシンプルです。
# day42_daily_scheduler.py
import time
from datetime import datetime, timedelta
def run_daily_job():
"""毎日決まった時間に実行したい処理の例です。"""
print("=== 日次バッチ処理開始 ===")
print("現在時刻:", datetime.now())
print("ここで日次レポート生成などを行います。")
print("=== 日次バッチ処理終了 ===\n")
def get_next_run_time(hour=9, minute=0):
"""次に実行すべき日時(今日または明日の指定時刻)を計算する関数です。"""
now = datetime.now()
# 今日の指定時刻(例:9:00)を作ります。
today_run = datetime(now.year, now.month, now.day, hour, minute)
if now < today_run:
# まだ今日の指定時刻前なら、今日の指定時刻が次の実行時刻です。
return today_run
else:
# すでに今日の指定時刻を過ぎているなら、翌日の指定時刻が次の実行時刻です。
next_day = now + timedelta(days=1)
return datetime(next_day.year, next_day.month, next_day.day, hour, minute)
def main():
print("=== 日次スケジューラを開始します ===")
print("毎日9:00にバッチ処理を実行します。")
while True:
next_run = get_next_run_time(hour=9, minute=0)
now = datetime.now()
# 次の実行時刻までの秒数を計算します。
wait_seconds = (next_run - now).total_seconds()
print("現在時刻:", now)
print("次の実行時刻:", next_run)
print("待機秒数:", wait_seconds)
# 次の実行時刻まで待機します。
time.sleep(wait_seconds)
# 待機が終わったらバッチ処理を実行します。
run_daily_job()
if __name__ == "__main__":
main()
Python重要ポイント:
- 「次にいつ実行すべきか」を
get_next_run_time()で計算している。 next_run - nowで「待つべき時間(timedelta)」を求めている。total_seconds()で「待つべき秒数」を取り出し、sleepに渡している。
このように、日時処理(Day 41)で学んだ datetime と timedelta が、定期実行のロジックの土台になる ことが分かります。
実務でよく使われる「OS側のスケジューラ」との役割分担
ここまでの例は「Pythonだけで定期実行を頑張る」形でしたが、 実務では次のような役割分担をすることが多いです。
- Pythonスクリプト
- 「1回分のバッチ処理」をきちんと書く(CSV集計、PDF生成など)
- OSのスケジューラ
- そのスクリプトを「いつ」「どの頻度で」実行するかを管理する
代表的なスケジューラ:
- Linux / macOS:
cron - Windows:タスクスケジューラ
イメージとしては、
- Python側:
python day39_csv_transform_tool.pyを1回実行すると、ちゃんと仕事をするように作る - OS側:そのコマンドを「毎日9時に実行するように設定する」
という分担です。
Day 42では、OSの設定方法の細かい手順までは踏み込みませんが、 「定期実行は、Pythonだけで完結させることもできるが、実務ではOSのスケジューラと組み合わせるのが一般的」 という考え方を持っておくと、後で応用しやすくなります。
定期実行の設計で意識したいポイント
1. 「1回分の処理」を必ず関数にまとめる
def run_daily_job():
# ここに処理を書く
...
Pythonこうしておくと、
- 手動でテストするとき:
run_daily_job()を直接呼べる - スケジューラから呼ぶとき:
main()の中でrun_daily_job()を呼ぶだけでよい
という形になり、コードが整理されます。
2. エラーが起きたときの振る舞いを考える
定期実行では、「たまたまその日だけデータが壊れていた」 などのケースも起こり得ます。
- エラーが起きたらログを残す
- エラーが起きてもスクリプト全体が落ちないように
try/exceptを使う - 次回の実行に影響が出ないようにする
などを意識すると、安定したバッチ処理になります。
簡単な例:
def run_daily_job():
print("=== 日次バッチ処理開始 ===")
try:
print("CSVを読み込みます...")
# ここで実際の処理を書く(例:ファイル読み込み・集計など)
print("処理が正常に完了しました。")
except Exception as e:
print("エラーが発生しました:", e)
# ここでログファイルに書き出すなども検討できます。
print("=== 日次バッチ処理終了 ===\n")
PythonDay 42ミニテンプレート:定期実行の「型」
最後に、Day 42の内容をまとめた「定期実行の型」を示します。
# day42_batch_template.py
import time
from datetime import datetime, timedelta
def run_batch():
"""定期的に実行したいバッチ処理をまとめた関数です。"""
print("=== バッチ処理開始 ===")
print("現在時刻:", datetime.now())
# ここに実際の処理を書く(CSV集計、PDF生成など)
print("ここで業務ロジックを実行します。")
print("=== バッチ処理終了 ===\n")
def get_next_run_time(hour=9, minute=0):
"""次の実行時刻を計算する関数です。"""
now = datetime.now()
today_run = datetime(now.year, now.month, now.day, hour, minute)
if now < today_run:
return today_run
else:
next_day = now + timedelta(days=1)
return datetime(next_day.year, next_day.month, next_day.day, hour, minute)
def scheduler_loop():
"""毎日指定時刻にバッチ処理を実行する簡易スケジューラです。"""
print("=== スケジューラ開始 ===")
print("毎日9:00にバッチ処理を実行します。")
while True:
next_run = get_next_run_time(hour=9, minute=0)
now = datetime.now()
wait_seconds = (next_run - now).total_seconds()
print("現在時刻:", now)
print("次の実行時刻:", next_run)
print("待機秒数:", wait_seconds)
time.sleep(wait_seconds)
run_batch()
def main():
scheduler_loop()
if __name__ == "__main__":
main()
Pythonこのテンプレートは、
- 「バッチ処理を1つの関数にまとめる」
- 「次の実行時刻を計算する」
- 「その時刻まで待ってから処理を実行する」
という、定期実行の基本的な考え方をそのままコードにしたものです。
Day 42のまとめ
Day 42では、定期実行として、
- バッチ処理とは「あるタイミングでまとめて一気に処理する」スタイルであること
- 定期実行とは「人が押す実行ボタンを、時間が押してくれる仕組み」であること
time.sleep()とwhile Trueを使った簡易的なスケジューラの作り方datetimeとtimedeltaを使って「次の実行時刻」を計算する考え方- 実務では、PythonスクリプトとOSのスケジューラ(cronやタスクスケジューラ)を組み合わせるのが一般的であること
- バッチ処理を関数にまとめ、エラー処理やログ出力を意識することで、安定した定期実行が実現できること
をステップバイステップで体験していただきました。
ここまで来ると、 「毎日・毎週・毎月の定型業務を、Pythonスクリプト+スケジューラで自動化する」 というイメージがかなり具体的になっているはずです。
次のステップでは、こうした定期実行の仕組みと、 これまで作ってきたCSV・Excel・PDF・pandasのツールを組み合わせて、 より実務に近い「業務自動化システム」を形にしていくことができます。
