PART 03 — バッチ処理設計書
バッチジョブ一覧・スケジュール・処理フロー・エラーハンドリング・再実行設計で定義すべきこと

"""
全バッチに共通するログ・エラーハンドリング・実行記録を提供する基底クラス。
各バッチはこのクラスを継承してexecute()メソッドを実装するだけでよい。
"""
import abc
import logging
import psycopg2
from dataclasses import dataclass
from datetime import datetime
from enum import Enum

logger = logging.getLogger(__name__)


class BatchStatus(str, Enum):
    RUNNING  = "RUNNING"
    SUCCESS  = "SUCCESS"
    FAILED   = "FAILED"
    SKIPPED  = "SKIPPED"


@dataclass
class BatchResult:
    status: BatchStatus
    processed_count: int = 0
    error_count: int = 0
    message: str = ""


class BaseBatch(abc.ABC):
    """バッチ基底クラス。全バッチはこのクラスを継承する。"""

    batch_id: str        # サブクラスで定義: 例 "B001"
    batch_name: str      # サブクラスで定義: 例 "在庫送信バッチ"

    def __init__(self, db_config: dict):
        self.db_config = db_config
        self.run_id: int | None = None
        self.started_at: datetime | None = None

    def run(self) -> BatchResult:
        """バッチ実行エントリーポイント（オーバーライド不可）"""
        self.started_at = datetime.now()
        logger.info(f"[{self.batch_id}] {self.batch_name} 開始")

        try:
            self._check_previous_run()
            self._record_start()
            result = self.execute()
            self._record_end(result)
            logger.info(f"[{self.batch_id}] 完了: {result}")
            return result

        except Exception as e:
            logger.exception(f"[{self.batch_id}] 異常終了: {e}")
            error_result = BatchResult(
                status=BatchStatus.FAILED,
                message=str(e)
            )
            self._record_end(error_result)
            self._send_alert(str(e))
            raise

    @abc.abstractmethod
    def execute(self) -> BatchResult:
        """バッチ本体処理。サブクラスで実装する。"""
        ...

    def _check_previous_run(self) -> None:
        """前回の実行が異常終了中のままでないか確認する"""
        with psycopg2.connect(**self.db_config) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "SELECT run_id FROM l_batch_log WHERE batch_id=%s AND status=%s LIMIT 1",
                    (self.batch_id, BatchStatus.RUNNING)
                )
                row = cur.fetchone()
                if row:
                    raise RuntimeError(
                        f"前回の実行（run_id={row[0]}）が RUNNING のままです。"
                        "手動でステータスを確認してください。"
                    )

    def _record_start(self) -> None:
        with psycopg2.connect(**self.db_config) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "INSERT INTO l_batch_log (batch_id, batch_name, status, started_at) "
                    "VALUES (%s, %s, %s, %s) RETURNING run_id",
                    (self.batch_id, self.batch_name, BatchStatus.RUNNING, self.started_at)
                )
                self.run_id = cur.fetchone()[0]
            conn.commit()

    def _record_end(self, result: BatchResult) -> None:
        if self.run_id is None:
            return
        with psycopg2.connect(**self.db_config) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "UPDATE l_batch_log SET status=%s, processed_count=%s, "
                    "error_count=%s, message=%s, ended_at=%s WHERE run_id=%s",
                    (result.status, result.processed_count, result.error_count,
                     result.message, datetime.now(), self.run_id)
                )
            conn.commit()

    def _send_alert(self, message: str) -> None:
        """エラー時のアラート送信（メール等）。必要に応じてオーバーライド。"""
        logger.error(f"ALERT: [{self.batch_id}] {message}")


# ── 在庫送信バッチの実装例 ─────────────────────────────
class StockSendBatch(BaseBatch):
    batch_id   = "B001"
    batch_name = "在庫送信バッチ"

    def execute(self) -> BatchResult:
        # 1. DBから更新在庫データを抽出
        # 2. CSV生成
        # 3. SFTP送信
        # ... 実際の処理を実装
        logger.info("在庫データを抽出中...")
        return BatchResult(status=BatchStatus.SUCCESS, processed_count=1500)