Python | 自動化：Python バッチ

Python
スポンサーリンク

概要（「決まった処理をまとめてやるスクリプト」がバッチ）

ここで言う「Pythonバッチ」は、
「人がボタンを押さなくても、決まった処理をまとめて実行するスクリプト」のことです。

例えば、毎朝こんなことを勝手にやってくれるものをイメージしてください。

データベースから前日のデータを取ってくる。
ExcelやCSVを読み込んで集計する。
結果をファイルに出力したり、メールで送る。
ログに成功・失敗を書き残す。

これを「1つのスクリプトとしてしっかり設計する」のが、Pythonバッチです。
cron やタスクスケジューラから呼ばれる前提で、エラーやログも含めて「放置しても回り続ける」ように作ります。

基本構成（main関数・設定・ログを最初に決める）

典型的な「Pythonバッチの骨組み」

最初に「雛形」を持っておくと、毎回ゼロから考えなくて済みます。
最低限、main関数・ログ出力・設定の3つを用意します。

# batch_example.py
import logging
from datetime import datetime
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent
LOG_FILE = BASE_DIR / "logs" / "batch_example.log"

def setup_logging():
    LOG_FILE.parent.mkdir(exist_ok=True)
    logging.basicConfig(
        filename=LOG_FILE,
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s"
    )

def main():
    logging.info("バッチ開始")
    try:
        run_job()
        logging.info("バッチ正常終了")
    except Exception as e:
        logging.exception(f"バッチ異常終了: {e}")
        raise

def run_job():
    logging.info("ここにメイン処理を書く")
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    logging.info(f"現在時刻: {now}")

if __name__ == "__main__":
    setup_logging()
    main()
Python

この骨組みで大事なのは、
「ログの場所を固定する」「main で try/except して必ずログに残す」ことです。
これだけで「夜中に失敗していたのに気づけない」という事故が一気に減ります。

実例１：CSVを毎日集計してレポートを出すバッチ

要件をシンプルに言語化する

例として、こんなバッチを作ってみます。

毎日、data/input/ に「日別売上CSV」が増えていく。
ファイル名は sales_YYYYMMDD.csv とする。
毎朝9時に「前日分のCSV」を読み込み、カテゴリ別売上を集計し、
output/report_YYYYMMDD.csv として保存する。

この要件を、そのままコードに落とします。

pandasを使った集計バッチの例

# daily_sales_batch.py
import logging
from datetime import datetime, timedelta
from pathlib import Path

import pandas as pd

BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data" / "input"
OUT_DIR = BASE_DIR / "output"
LOG_FILE = BASE_DIR / "logs" / "daily_sales_batch.log"

def setup_logging():
    LOG_FILE.parent.mkdir(exist_ok=True)
    logging.basicConfig(
        filename=LOG_FILE,
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s"
    )

def get_target_date(today=None):
    if today is None:
        today = datetime.today()
    return (today - timedelta(days=1)).date()

def build_input_path(target_date):
    fname = f"sales_{target_date.strftime('%Y%m%d')}.csv"
    return DATA_DIR / fname

def build_output_path(target_date):
    fname = f"report_{target_date.strftime('%Y%m%d')}.csv"
    return OUT_DIR / fname

def run_job():
    OUT_DIR.mkdir(exist_ok=True)
    target_date = get_target_date()
    logging.info(f"対象日: {target_date}")

    input_path = build_input_path(target_date)
    output_path = build_output_path(target_date)

    if not input_path.exists():
        logging.warning(f"入力ファイルがありません: {input_path}")
        return

    logging.info(f"入力ファイル読み込み: {input_path}")
    df = pd.read_csv(input_path)

    required_cols = {"date", "category", "amount"}
    if not required_cols.issubset(df.columns):
        missing = required_cols - set(df.columns)
        raise ValueError(f"必要な列が足りません: {missing}")

    df["amount"] = pd.to_numeric(df["amount"], errors="coerce")

    summary = (
        df.groupby("category", as_index=False)["amount"]
        .sum()
        .rename(columns={"amount": "total_amount"})
    )

    logging.info(f"出力ファイル保存: {output_path}")
    summary.to_csv(output_path, index=False, encoding="utf-8-sig")

def main():
    logging.info("日次売上バッチ開始")
    try:
        run_job()
        logging.info("日次売上バッチ正常終了")
    except Exception as e:
        logging.exception(f"日次売上バッチ異常終了: {e}")
        raise

if __name__ == "__main__":
    setup_logging()
    main()
Python

ここで深掘りしたいポイントは4つあります。

一つ目は「対象日」を関数で切り出していることです。
本番は「昨日」ですが、テストしたいときに任意の日付を渡せる設計にしておくと、後から楽になります。

二つ目は「パスを組み立てる関数」を作っていることです。
ファイル名の規則が変わっても、その関数だけ直せばよくなります。

三つ目は「入力ファイルが無いときの扱い」を意図的に決めていることです。
この例では警告ログだけ出して終了にしていますが、
「必ずある前提ならエラーで止める」という設計もありです。

四つ目は「列チェック」と「型変換」を明示的に行っていることです。
テンプレが壊れていても気づけるように、「必須列が無ければ即例外」を投げています。

実例２：APIを叩いてデータ取得 → DB保存するバッチ

外部サービスと連携するバッチのイメージ

もう少し“バッチ感”の強い例を出します。

外部API（例えばWebサービスのレポートAPI）からデータを取ってくる。
JSONを整形して、ローカルのSQLiteやPostgreSQLへ保存する。
毎時間実行して、差分を蓄積していく。

この場合も、「やること」は同じです。

設定を読み込む。
ログを初期化する。
APIを呼ぶ。
レスポンスを検証する。
DBに保存する。

これを一つの流れとして main にまとめます。

超シンプルなAPIバッチの骨組み

# fetch_api_batch.py
import logging
from datetime import datetime
from pathlib import Path

import requests
import sqlite3

BASE_DIR = Path(__file__).resolve().parent
LOG_FILE = BASE_DIR / "logs" / "fetch_api_batch.log"
DB_FILE = BASE_DIR / "data" / "api_data.db"

API_URL = "https://example.com/api/data"

def setup_logging():
    LOG_FILE.parent.mkdir(exist_ok=True)
    logging.basicConfig(
        filename=LOG_FILE,
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s"
    )

def get_db_connection():
    DB_FILE.parent.mkdir(exist_ok=True)
    conn = sqlite3.connect(DB_FILE)
    return conn

def init_db(conn):
    conn.execute("""
        CREATE TABLE IF NOT EXISTS api_data (
            id INTEGER PRIMARY KEY,
            value TEXT,
            fetched_at TEXT
        )
    """)
    conn.commit()

def fetch_data():
    logging.info(f"API呼び出し: {API_URL}")
    resp = requests.get(API_URL, timeout=10)
    resp.raise_for_status()
    return resp.json()

def save_data(conn, data):
    now = datetime.utcnow().isoformat()
    conn.execute(
        "INSERT INTO api_data (value, fetched_at) VALUES (?, ?)",
        (str(data), now)
    )
    conn.commit()

def run_job():
    conn = get_db_connection()
    try:
        init_db(conn)
        data = fetch_data()
        save_data(conn, data)
        logging.info("データ保存完了")
    finally:
        conn.close()

def main():
    logging.info("API取得バッチ開始")
    try:
        run_job()
        logging.info("API取得バッチ正常終了")
    except Exception as e:
        logging.exception(f"API取得バッチ異常終了: {e}")
        raise

if __name__ == "__main__":
    setup_logging()
    main()
Python

ここで重要なのは、「ネットワークエラーやAPIエラーを logging.exception で必ずログに落とす」ことと、「DB接続のクローズを finally で保証する」ことです。
バッチは失敗すること自体は避けられませんが、「失敗の痕跡が残るか」が運用での生き死にを分けます。

設計の重要ポイントを深掘り（堅いバッチにするための考え方）

絶対パスとPathで「どこから実行しても動く」ようにする

バッチは cron やタスクスケジューラから実行されることが多く、
カレントディレクトリがどこか分からない状態で動きます。

open("input.csv") のような相対パスは、
実行環境によって簡単に壊れます。

__file__ から「スクリプトが置かれているディレクトリ」を基準に、
Pathで絶対パスを組み立てる癖をつけると、どこから実行しても動くようになります。

BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data"
Python

この書き方を「バッチの雛形」にしておくと、後から自分を助けてくれます。

ログレベルとメッセージ粒度

loggingには DEBUG / INFO / WARNING / ERROR / CRITICAL があります。
バッチでは、普段は INFO レベルで十分ですが、
開発中は DEBUG まで出したくなることもあります。

最初から format やレベルを一箇所で設定し、
「何をINFOにするか」「何をWARNINGにするか」を決めておくと、
ログファイルが読みやすくなります。

大事なのは、「誰が見ても、いつ、何を、どこまでやったか分かるメッセージ」にすることです。
「処理開始」「入力ファイル名」「件数」「処理終了」「異常終了の理由」
このあたりを書く癖をつければ、運用のストレスがかなり減ります。

失敗したときに再実行しやすい設計（冪等性）

バッチは途中で失敗することがあります。
例えば、半分までDBに書いたところでエラーになった、とか。
そのときに「もう一度実行してもおかしくならない」ように設計しておくと、運用が楽です。

例えば、こんな工夫があります。

対象データを「日付やID」単位で完全に上書きする。
処理前に既存データを削除してから入れ直す。
一度処理したファイルを「done」フォルダへ移動する。
処理済みフラグをDBに持っておく。

「再実行したときに何が起きるか」を最初に想像しながら、
データの持ち方を決めると、後から「変な二重登録」に悩まされにくくなります。

バッチとcron / タスクスケジューラのつなぎこみ

Linux / Macでの定期実行イメージ

先ほどの daily_sales_batch.py を毎朝9時に動かしたい場合は、
cronに次のような行を登録します。

0 9 * * * /home/you/project/venv/bin/python /home/you/project/daily_sales_batch.py >> /home/you/project/logs/cron_stdout.log 2>&1

「どのPythonで」「どのスクリプトを」「どのログに出すか」を、
バッチ側ではなく cron 側で決めるイメージです。

Windowsでの定期実行イメージ

タスクスケジューラでは、

プログラム／スクリプトに
C:\project\venv\Scripts\python.exe

引数に
C:\project\daily_sales_batch.py

開始フォルダに
C:\project

を指定し、トリガーを「毎日9:00」にする、という感じです。

重要なのは、「スクリプトは普通のPythonとして書いて、スケジュールはOSが担当する」という分担をきっちり分けることです。
バッチ側に「時間管理」を書かない方がすっきりします。

まとめ（「雛形を決める」「ログとパスをきちんと設計する」だけで一気に“バッチらしく”なる）

Pythonバッチの本質は、特別な技術ではなく「きちんとしたスクリプトの書き方」です。

main関数を作り、ログを初期化し、絶対パスでファイルを扱い、
エラー時には logging.exception で理由を残す。
対象日やファイル名の規則を関数に切り出し、
再実行したときに壊れないようなデータ設計をする。

この型を一度作ってしまえば、後は「中身の処理」を入れ替えるだけで
いろいろなバッチを作れるようになります。

スポンサーリンク
タイトルとURLをコピーしました