2026年9月14日 – AIに仕事を任せたい人のためのPython塾 ― Manage AI ―

ちょっとしたスクリプトは「動けば良い」となりがちで、運用に回すとログが足りなかったり、設定がハードコーディングされて苦労したりします。本記事では、小規模なAI／データ処理を現場で安全に運用するためのCLI設計とログ設計を、Python 3.9以降の標準ライブラリだけで動かせるテンプレートとチェックリストでまとめます。コードはWordPressに貼り付けられる<pre><code>形式で掲載します。

1）なぜ小さなCLIツールと良いログ設計が必要か（現場の失敗例）

現場でよくあるつまずきと影響を整理します。

よくある問題	現場での影響
引数が曖昧でHelpが無い	使う人が誤操作しやすく、自動化に組み込みにくい
設定がコード内にハードコーディング	環境ごとに編集が必要で漏洩リスクがある
ログがプレーンテキストで機械処理しにくい	障害検出や集約が難しく、原因追跡が遅れる
エラーが標準出力に散らばる	例外時の再現性が低く、対応が困難になる

2）要件整理：引数・設定・出力・監視

設計を始める前に満たすべき要件を明確にします。

領域	要件（実務的観点）
引数	必須・任意を明確にし、サブコマンドで処理を分離する。`-h/--help`を充実させる
設定	設定ファイル（configparser）を環境変数で上書きする。秘密情報は設定ファイルに保存しない
出力／ログ	コンソールは人間向け、ファイルはJSON Lines形式で保存し、ローテーションを設定する
監視	相関ID、ログレベル、処理時間、エラーを検索できるようにする

3）argparseで作る使いやすいCLI

完成テンプレートでは、入力ファイルの空行を除いた行数を数え、結果をJSONファイルへ保存するprocessサブコマンドを用意します。--inputと--outputは必須です。

グローバルオプションの--configと--correlation-idは、次のようにサブコマンドより前へ指定します。

python cli_tool.py --config config.ini --correlation-id demo-001 process --input input.txt --output result.json

ヘルプは以下のコマンドで確認できます。

python cli_tool.py -h
python cli_tool.py process -h

必須引数が不足している場合など、引数解析のエラーはargparseが標準エラー出力へ表示し、終了コード2で終了します。

4）configparserと環境変数で安全に設定を読み込む手順

ログファイルの保存先とログレベルをINIファイルで管理します。たとえば、config.iniを次の内容で作成します。

[app]
log_file = logs/cli.jsonl
log_level = INFO

完成コードでは、APP_LOG_FILEとAPP_LOG_LEVELが設定されていれば、INIファイルの値より環境変数を優先します。設定ファイルを指定しない場合は、コード内の非機密な初期値を使用します。

APP_LOG_LEVEL=DEBUG python cli_tool.py --config config.ini process --input input.txt --output result.json

APIキーなどの秘密情報はINIファイルへ書かず、実際にAPI処理を追加する段階でos.environまたはos.getenv()から直接取得してください。このテンプレート自体は外部APIを呼び出さないため、APIキーは要求しません。

5）logging入門：ハンドラとフォーマッタ（プレーン／JSON）

ポイントは「コンソールは人間向け、ファイルは機械処理向け」の使い分けです。完成コードでは、ルートロガーに次の2つのハンドラを設定します。

出力先	設定
コンソール	`StreamHandler`を使用し、INFO以上を読みやすいプレーンテキストで出力
ログファイル	`RotatingFileHandler`と独自のJSONフォーマッタを使用し、1行1JSONで出力

ログファイル側のレベルはlog_levelで変更できます。初期値はINFOです。DEBUGを指定した場合でも、コンソールはINFO以上、ファイルはDEBUG以上という使い分けになります。

6）実践：ログ回転、例外の一元キャッチ、相関IDの付与

完成コードでは、ログファイルが5MBに達したらローテーションし、過去5ファイルを保持します。maxBytesとbackupCountは運用条件に合わせて調整してください。

相関IDはcontextvars.ContextVarに保存します。--correlation-idが指定されていない場合は、CLIの実行ごとにUUIDを生成します。同じ実行中に出力されたログを、共通のcorrelation_idで検索できます。

コマンド処理中の例外はmain()で一元的に捕捉し、logger.exception()でスタックトレースを記録して終了コード1を返します。Ctrl+CなどのKeyboardInterruptは終了コード130で終了します。

7）入出力とJSONログの確認例

input.txtを次の内容で作成します。

alpha

beta

設定ファイルを使って実行します。

python cli_tool.py --config config.ini --correlation-id demo-001 process --input input.txt --output result.json

result.jsonには次の結果が保存されます。

{
  "input": "input.txt",
  "non_empty_lines": 2,
  "correlation_id": "demo-001"
}

logs/cli.jsonlには、開始、処理完了、コマンド成功のログが1行1JSONで記録されます。時刻や処理時間は実行ごとに変わります。

{"timestamp":"2025-01-01T00:00:00+00:00","level":"INFO","logger":"cli_tool","message":"processing_finished","correlation_id":"demo-001","duration_ms":1.25,"input_path":"input.txt","output_path":"result.json","non_empty_lines":2}

入力ファイルが存在しない場合は、例外情報を含むERRORログが出力され、コマンドは終了コード1で終了します。

8）モニタリング連携の入口（ログからSLO／アラートにつなげる方法）

まずはログ構造を整え、次のチェックリストに従ってください。

目的	短いチェックリスト
エラー検知	JSONログの`level`、`correlation_id`、`exception`を検索できるか確認する
パフォーマンス監視	`processing_finished`の`duration_ms`を集計し、平均値や上位値を確認する
アラート設計	短期のエラーバーストと、長期のエラーレート上昇を別の条件として扱う

9）デプロイと運用チェックリスト

CIから本番運用までの最低限の確認項目です。

カテゴリ	チェック項目
CI	必要に応じてflake8やBlackによるチェックを行い、主要フローと異常系をテストする
設定	秘密情報は環境変数などで注入し、設定ファイルには保存しない。本番の設定ファイルは不用意に変更できない権限にする
実行	終了コードを監視し、systemdやコンテナを利用する場合は再起動条件とログ転送先を設定する
ログ	ローテーション後のファイル数と容量を確認し、必要な保存期間に合わせて調整する

10）完成コード（そのまま貼れる短縮テンプレート）

以下をcli_tool.pyとして保存してください。CLI、設定読み込み、環境変数による上書き、JSONログ、ログ回転、相関ID、例外処理、入出力処理を含むテンプレートです。

#!/usr/bin/env python3
import argparse
import json
import logging
import os
import sys
import time
import uuid
from configparser import ConfigParser
from contextvars import ContextVar
from datetime import datetime, timezone
from logging.handlers import RotatingFileHandler
from pathlib import Path


correlation_id_var = ContextVar("correlation_id", default="-")
logger = logging.getLogger("cli_tool")


class JsonFormatter(logging.Formatter):
    """logging.LogRecordを1行のJSONへ変換する。"""

    def format(self, record):
        payload = {
            "timestamp": datetime.fromtimestamp(
                record.created, tz=timezone.utc
            ).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "correlation_id": correlation_id_var.get(),
        }

        for key in (
            "duration_ms",
            "input_path",
            "output_path",
            "non_empty_lines",
        ):
            value = getattr(record, key, None)
            if value is not None:
                payload[key] = value

        if record.exc_info:
            payload["exception"] = self.formatException(record.exc_info)

        return json.dumps(payload, ensure_ascii=False)


def parse_args():
    parser = argparse.ArgumentParser(
        description="入力ファイルを処理し、結果と運用ログを出力します。"
    )
    parser.add_argument(
        "--config",
        type=Path,
        help="INI形式の設定ファイル",
    )
    parser.add_argument(
        "--correlation-id",
        help="ログを関連付ける相関ID。省略時はUUIDを生成します。",
    )

    subparsers = parser.add_subparsers(
        dest="command",
        required=True,
    )

    process_parser = subparsers.add_parser(
        "process",
        help="入力ファイルの空行を除いた行数を集計します。",
    )
    process_parser.add_argument(
        "--input",
        type=Path,
        required=True,
        help="UTF-8の入力テキストファイル",
    )
    process_parser.add_argument(
        "--output",
        type=Path,
        required=True,
        help="結果を書き込むJSONファイル",
    )
    process_parser.set_defaults(handler=run_process)

    return parser.parse_args()


def load_config(config_path):
    parser = ConfigParser()
    parser.read_dict(
        {
            "app": {
                "log_file": "logs/cli.jsonl",
                "log_level": "INFO",
            }
        }
    )

    if config_path is not None:
        if not config_path.is_file():
            raise FileNotFoundError(
                f"設定ファイルが見つかりません: {config_path}"
            )
        with config_path.open("r", encoding="utf-8") as file:
            parser.read_file(file)

    return {
        "log_file": os.getenv(
            "APP_LOG_FILE",
            parser.get("app", "log_file"),
        ),
        "log_level": os.getenv(
            "APP_LOG_LEVEL",
            parser.get("app", "log_level"),
        ),
    }


def get_log_level(level_name):
    level = getattr(logging, level_name.upper(), None)
    if not isinstance(level, int):
        raise ValueError(f"不正なログレベルです: {level_name}")
    return level


def setup_logging(log_file, log_level):
    file_level = get_log_level(log_level)
    log_path = Path(log_file)
    log_path.parent.mkdir(parents=True, exist_ok=True)

    root_logger = logging.getLogger()
    root_logger.setLevel(logging.DEBUG)

    for handler in root_logger.handlers[:]:
        root_logger.removeHandler(handler)
        handler.close()

    console_handler = logging.StreamHandler(sys.stderr)
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(
        logging.Formatter(
            "%(asctime)s %(levelname)s "
            "correlation_id=%(correlation_id)s %(message)s"
        )
    )
    console_handler.addFilter(CorrelationIdFilter())

    file_handler = RotatingFileHandler(
        log_path,
        maxBytes=5 * 1024 * 1024,
        backupCount=5,
        encoding="utf-8",
    )
    file_handler.setLevel(file_level)
    file_handler.setFormatter(JsonFormatter())

    root_logger.addHandler(console_handler)
    root_logger.addHandler(file_handler)


class CorrelationIdFilter(logging.Filter):
    def filter(self, record):
        record.correlation_id = correlation_id_var.get()
        return True


def setup_fallback_logging():
    """設定読み込み前の例外を標準エラー出力へ記録する。"""
    root_logger = logging.getLogger()
    if root_logger.handlers:
        return

    root_logger.setLevel(logging.INFO)
    handler = logging.StreamHandler(sys.stderr)
    handler.setFormatter(
        logging.Formatter(
            "%(asctime)s %(levelname)s "
            "correlation_id=%(correlation_id)s %(message)s"
        )
    )
    handler.addFilter(CorrelationIdFilter())
    root_logger.addHandler(handler)


def run_process(args):
    started_at = time.perf_counter()

    logger.info(
        "processing_started",
        extra={
            "input_path": str(args.input),
            "output_path": str(args.output),
        },
    )

    text = args.input.read_text(encoding="utf-8")
    non_empty_lines = sum(
        1 for line in text.splitlines() if line.strip()
    )

    result = {
        "input": str(args.input),
        "non_empty_lines": non_empty_lines,
        "correlation_id": correlation_id_var.get(),
    }

    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(
        json.dumps(result, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )

    duration_ms = round(
        (time.perf_counter() - started_at) * 1000,
        2,
    )
    logger.info(
        "processing_finished",
        extra={
            "duration_ms": duration_ms,
            "input_path": str(args.input),
            "output_path": str(args.output),
            "non_empty_lines": non_empty_lines,
        },
    )


def main():
    args = parse_args()
    correlation_id = args.correlation_id or str(uuid.uuid4())
    token = correlation_id_var.set(correlation_id)

    try:
        config = load_config(args.config)
        setup_logging(
            config["log_file"],
            config["log_level"],
        )

        logger.info("command_started")
        args.handler(args)
        logger.info("command_succeeded")
        return 0
    except KeyboardInterrupt:
        setup_fallback_logging()
        logger.warning("command_interrupted")
        return 130
    except Exception:
        setup_fallback_logging()
        logger.exception("command_failed")
        return 1
    finally:
        correlation_id_var.reset(token)


if __name__ == "__main__":
    sys.exit(main())

11）次の一歩

まずは上記コード、config.ini、input.txtを同じディレクトリに保存し、記載したコマンドで実行してください。その後、正常系と入力ファイルが存在しない異常系の両方を試し、result.json、コンソール出力、logs/cli.jsonl、終了コードを確認します。

ログを一定期間保存して検索できることを確認したら、必要に応じてログ集約基盤やメトリクス監視へつなげます。外部ライブラリを導入する前に、まずは標準ライブラリで引数、設定、ログ、例外、終了コードの一貫した設計を作ることが重要です。

まとめ

CLIはargparseで設計し、サブコマンド、必須引数、-h/--helpを用意する。
非機密設定はconfigparserで管理し、環境変数による上書きを可能にする。
ログはコンソールとJSONファイルを使い分け、ローテーションと相関IDを設定する。
例外を一元的に記録し、自動実行側が判定できる終了コードを返す。
処理時間やエラーをJSONログから集計し、監視やアラートへつなげる。

小さな自動化スクリプトほど、運用方法を最初に整えることで効果が持続します。まずはテンプレートを実行し、現場の保存期間や監視条件に合わせて調整してください。

月	火	水	木	金	土	日
	1	2	3	4	5	6
7	8	9	10	11	12	13
14	15	16	17	18	19	20
21	22	23	24	25	26	27
28	29	30