Python Logging Best Practices

Self-Evolving Skill: このスキルは使用を通じて改善されます。指示が誤っている、パラメータが変動している、回避策が必要だった場合 — このファイルを即座に修正してください。延期しないでください。実際に再現可能な問題の場合のみ更新します。

このスキルを使用する場面

以下の場合にこのスキルを使用してください:

• 任意のサービスまたはスクリプトの Python logging をセットアップする場合
• 分析用に構造化 JSONL logging を設定する場合
• ログローテーションを実装する場合
• 軽量（無依存）とフル機能の logging の間で選択する場合
• コンテナ化、systemd、またはローカルアプリケーションに logging を追加する場合

概要

機械読取性（Claude Code 分析）と運用信頼性に最適化された Python logging パターンの統一リファレンス。最も軽量な実行可能なアプローチで開始し、必要な場合のみスケールアップします。

決定ヒューリスティック: 軽量で開始、必要に応じてスケールアップ

単一マシン上の < 5 サービス、< 1 イベント/秒ですか？
  はい → 軽量パターン（print + JSONL テレメトリ）
  いいえ → コンテナ化 / サーバーレスですか？
    はい → stdout JSON（任意のライブラリ）、ファイルローテーションなし
    いいえ → OTel トレースが必要ですか？
      はい → structlog + OTel
      いいえ → loguru（CLI ツール）または stdlib RotatingFileHandler

アプローチ	ユースケース	利点	欠点
Lightweight	小規模 systemd サービス、自社ホスト、単一オペレータ	無依存、journald 統合、最小限のコード	重大度フィルタリングなし、モジュール単位の制御なし
loguru	CLI ツール、スクリプト、ローカルサービス	ゼロ設定、組み込みローテーション、優れた DX	外部依存、真のスキーマ強制なし
structlog	本番サービス、OTel 統合	ContextVars、プロセッサチェーン、OTel ネイティブ	学習曲線が急
stdlib	LaunchAgent デーモン、無依存制約	依存関係なし、Python 3.13 merge_extra	より多くのボイラープレート、構造化デフォルトなし
Logfire	AI/LLM オブザーバビリティ、Pydantic アプリ	OTel ベース、トークン/コスト追跡、SQL	SaaS 依存、新しいエコシステーム

---

推奨: 軽量パターン（無依存）

対象: < 5 systemd サービス、単一サーバー、単一オペレータ。ccmax-monitor によって本番環境で実証済み。

このパターンは 2チャネルアーキテクチャ を使用します:

• チャネル 1: print(flush=True) → systemd journald（運用ログ、人間が読める形式）
• チャネル 2: アペンドオンリー JSONL ファイル（構造化テレメトリ、機械が読める形式）

これは 12-Factor App の「ログをイベントストリームとして扱う」原則にマップされます。journald は ops（ローテーション、フィルタリング、メタデータ）を処理し、JSONL ファイルは事後分析用のドメインテレメトリとして機能します。

アーキテクチャ: 3 つの関心の分離

関心	メカニズム	目的	ライフサイクル
Ops logging	print() → journald	人間によるデバッグ、journalctl -u service -f	journald により管理（自動ローテーション）
Telemetry	JSONL ファイル（telemetry.jsonl）	構造化監査証跡、AI/LLM 分析	アペンドオンリー、サイズベースでローテーション
State recovery	WAL ファイル（オプション）	元に戻せない操作のクラッシュ回復	一時的、成功時に削除

完全な軽量パターンの例

"""アペンドオンリー JSONL テレメトリロガーとサイズベースのローテーション。

外部依存なし。ops logging 用の systemd journald と
構造化機械可読テレメトリ用の別々の JSONL ファイルで動作します。
"""

import json
from datetime import datetime, timezone
from pathlib import Path

TELEMETRY_PATH = Path(__file__).parent / "telemetry.jsonl"
MAX_SIZE = 10 * 1024 * 1024  # 10 MB
BACKUP_COUNT = 3             # ローテーション済みバックアップを 3 つ保持（合計〜30MB）


def log_event(event_type: str, data: dict) -> None:
    """構造化 JSON 行を telemetry.jsonl に追加します。"""
    entry = {
        "ts": datetime.now(timezone.utc).isoformat(),
        "type": event_type,
        **data,
    }
    line = json.dumps(entry, separators=(",", ":")) + "\n"

    try:
        try:
            if TELEMETRY_PATH.stat().st_size > MAX_SIZE:
                _rotate()
        except FileNotFoundError:
            pass

        with open(TELEMETRY_PATH, "a") as f:
            f.write(line)
    except OSError as e:
        # stderr へのフォールバック（journald によってキャプチャ）
        print(f"[telemetry] write failed: {e}", file=__import__("sys").stderr, flush=True)


def _rotate() -> None:
    """テレメトリファイルをローテーション: .jsonl → .jsonl.1 → .jsonl.2 → .jsonl.3"""
    for i in range(BACKUP_COUNT, 1, -1):
        src = TELEMETRY_PATH.with_suffix(f".jsonl.{i - 1}")
        dst = TELEMETRY_PATH.with_suffix(f".jsonl.{i}")
        if src.exists():
            dst.unlink(missing_ok=True)
            src.rename(dst)
    backup = TELEMETRY_PATH.with_suffix(".jsonl.1")
    backup.unlink(missing_ok=True)
    TELEMETRY_PATH.rename(backup)


# === Ops logging（stdout 経由で journald に送信） ===

def log(msg: str) -> None:
    """人間が読める運用ログ行。journald によってキャプチャされます。"""
    ts = datetime.now(timezone.utc).strftime("%H:%M:%S")
    print(f"[{ts}] {msg}", flush=True)

使用例:

# 運用ログ（人間が journalctl -u myservice -f で読む）
log("Refreshing token for account X")
log("Switch: account A → account B (reason: 5h breach)")

# テレメトリ（機械が jq/DuckDB/Claude Code で読む）
log_event("token_refresh", {"account": "X", "expires_in_h": 8.0, "token_fp": "abc12345"})
log_event("account_switch", {"from": "A", "to": "B", "reason": "5h_breach"})

セキュリティ: トークンフィンガープリント（正規表現リダクションではなく）

シークレットをログパイプラインに渡さないでください。 非可逆的なフラグメントのみをログに記録します:

def _token_fingerprint(token: str) -> str:
    """トークンの中間セクションから一意に識別可能な文字を抽出します。

    プレフィックス（sk-ant-oat01-）とサフィックス（...AA）は
    トークン間で共通しています。
    プレフィックス後の文字 14-22 がトークンごとに最も一意です。
    中間スライスは、プレフィックスベースのアプローチが公開する
    タイププレフィックスメタデータのリークを回避します。
    """
    if len(token) > 25:
        return token[14:22]
    return token[:8] if token else ""

# 使用例: フィンガープリントをログに記録し、決してトークンを渡さない
log_event("token_refresh", {"account": name, "token_fp": _token_fingerprint(token)})

これが正規表現リダクションフィルターより優れている理由:

アプローチ	セキュリティ	メンテナンス	故障モード
トークンフィンガープリント（スライスのみをログに記録）	シークレットはログパイプラインに入らない	ゼロ — 任意のトークン形式で動作	失敗できない — リダクションするものがない
正規表現リダクションフィルター	シークレットはパスされ、出力時にフィルタリング	新しいトークン形式用に正規表現を更新する必要あり	サイレント失敗 = ログにシークレット

これは OWASP Logging Cheat Sheet に一致しています: 「ログエントリにセンシティブデータが含まれていないことを確認してください。」主要プラットフォーム（AWS、Stripe、GitHub）は、完全なトークンと正規表現スクラビングではなく、別々の非シークレット識別子または部分的なトークン表示を使用しています。

正規表現フィルターは 多層防御のバックストップ として有用なままです。主要なコントロールではなく。

オブザーバビリティとしてのヘルスエンドポイント

小規模デプロイの場合、豊富な JSON ヘルスエンドポイントはログ集約を置き換えます:

@app.get("/api/status")
def status():
    """ホワイトボックスモニタリング — オンデマンドの現在の状態。"""
    return {"active_account": ..., "accounts": [...], "polled_at": ...}

@app.get("/api/vault-health")
def vault_health():
    """すべてのアカウントのトークンヘルス。"""
    return {name: {"status": "healthy", "expires_in": "7.5h", ...} for ...}

これは ヘルスエンドポイントモニタリングパターン（Microsoft Azure Architecture Center）/ ヘルスチェック API パターン（microservices.io）です。ダッシュボード IS モニタリングツール — Grafana/Prometheus は不要です。

サービス自体が構造化 JSON として独自の運用状態を提供する場合、以下が得られます:

• リアルタイム 現在の状態（ログ取り込みパイプラインによる遅延なし）
• ゼロインフラストラクチャ（ログシッパー、ストレージ、またはクエリエンジンなし）
• AI が解析可能（Claude Code は直接 curl して分析できます）

FOSS CLI ツールを使用した事後分析

ログ集約スタック不要。これらのシングルバイナリツールは JSONL で直接動作します:

# DuckDB — JSONL での SQL 分析（最も強力）
duckdb -c "SELECT type, count(*) FROM read_json_auto('telemetry.jsonl') GROUP BY 1 ORDER BY 2 DESC"

# jq — アドホック JSON フィルタリング
jq 'select(.type == "token_refresh")' telemetry.jsonl

# journalctl — 既に JSONL をネイティブにエクスポート
journalctl -u ccmax-switcher -o json --since "1h ago" | jq 'select(.PRIORITY == "3")'

# lnav — SQL を備えた対話型ターミナルログビューア
lnav telemetry.jsonl

# llm（Simon Willison）— LLM にパイプして AI 事後分析
journalctl -u myservice --since "2h ago" --priority=err -o json | llm "analyze root cause"

軽量を超えてアップグレードする場合

以下のいずれかが当てはまる場合は loguru/structlog にアップグレードしてください:

• 複数ホスト上の > 5 サービス（トレース ID が相関に必要）
• > 10 イベント/秒 継続（非同期シンク、orjson が必要）
• 複数オペレータ がモジュール単位のログレベルフィルタリングを必要とする
• コンプライアンス要件 が署名付きの構造化監査証跡を要求する
• コンテナ/K8s デプロイメント（stdout JSON が標準）

---

フル機能: Loguru + JSONL パターン

logging ライブラリの恩恵を受ける CLI ツール、スクリプト、サービス用:

ログローテーション（ローカル/CLI アプリの場合は常に設定）

from loguru import logger

logger.add(
    log_path,
    rotation="10 MB",
    retention="7 days",
    compression="gz"
)

# stdlib 代替案（無依存）
from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler(
    log_path,
    maxBytes=100 * 1024 * 1024,  # 100MB
    backupCount=5
)

コンテナ/サーバーレスアプリ: ファイルローテーションはスキップしてください。stdout/stderr を JSON として ログに記録してください。コンテナランタイムが収集とローテーションを処理します。

JSONL フォーマット（機械が読める）

# 行あたり 1 つの JSON オブジェクト - jq で解析可能
{"timestamp": "2026-01-14T12:45:23.456Z", "level": "info", "message": "..."}

ファイル拡張子: 常に .jsonl を使用（.json または .log ではなく）

パフォーマンス: > 10k レコード/秒の場合は json.dumps() の代わりに orjson を使用:

import orjson

def json_formatter(record) -> str:
    log_entry = { ... }
    return orjson.dumps(log_entry).decode()

正規表現リダクション（多層防御）

トークンフィンガープリント とともに バックストップ として使用します。主要なコントロールではなく:

import re

REDACT_PATTERNS = [
    (re.compile(r'AKIA[0-9A-Z]{16}'), '[REDACTED_AWS_KEY]'),
    (re.compile(r'sk-[a-zA-Z0-9]{48}'), '[REDACTED_API_KEY]'),
    (re.compile(r'(?i)bearer\s+[a-zA-Z0-9._~+/=-]+'), '[REDACTED_BEARER]'),
]

def redact_filter(record):
    for pattern, replacement in REDACT_PATTERNS:
        record["message"] = pattern.sub(replacement, record["message"])
    return True

logger.add(sink, filter=redact_filter)

シャットダウン — 常にキューに入ったメッセージをフラッシュ

import asyncio
from loguru import logger

async def main():
    logger.add("app.jsonl", enqueue=True)
    await logger.complete()

asyncio.run(main())
# 同期: logger.remove()

完全な Loguru + JSONL の例

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.13"
# dependencies = ["loguru", "orjson"]
# ///

import re
import sys
from pathlib import Path
from uuid import uuid4

import orjson
from loguru import logger

REDACT_PATTERNS = [
    (re.compile(r'AKIA[0-9A-Z]{16}'), '[REDACTED_AWS_KEY]'),
    (re.compile(r'sk-[a-zA-Z0-9]{48}'), '[REDACTED_API_KEY]'),
]


def json_formatter(record) -> str:
    log_entry = {
        "timestamp": record["time"].strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
        "level": record["level"].name.lower(),
        "component": record["function"],
        "operation": record["extra"].get("operation", "unknown"),
        "operation_status": record["extra"].get("status", None),
        "trace_id": record["extra"].get("trace_id"),
        "message": record["message"],
        "context": {k: v for k, v in record["extra"].items()
                   if k not in ("operation", "status", "trace_id", "metrics")},
        "metrics": record["extra"].get("metrics", {}),
        "error": None
    }

    if record["exception"]:
        exc_type, exc_value, _ = record["exception"]
        log_entry["error"] = {
            "type": exc_type.__name__ if exc_type else "Unknown",
            "message": str(exc_value) if exc_value else "Unknown error",
        }

    return orjson.dumps(log_entry).decode()


def redact_filter(record):
    for pattern, replacement in REDACT_PATTERNS:
        record["message"] = pattern.sub(replacement, record["message"])
    return True


def setup_logger(app_name: str, log_dir: Path | None = None):
    logger.remove()
    logger.add(sys.stderr, format=json_formatter, filter=redact_filter, level="INFO")
    if log_dir is not None:
        log_dir.mkdir(parents=True, exist_ok=True)
        logger.add(
            str(log_dir / f"{app_name}.jsonl"),
            format=json_formatter,
            filter=redact_filter,
            rotation="10 MB",
            retention="7 days",
            compression="gz",
            level="DEBUG"
        )
    return logger

セマンティックフィールドリファレンス

フィールド	型	目的
timestamp / ts	ISO 8601	イベント順序付け（ミリ秒精度最小）
level / type	string	重大度またはイベントタイプ
component / svc	string	モジュール、関数、またはサービス名
operation	string	実行中のアクション
operation_status	string	started/success/failed/skipped
trace_id	UUID4 or OTel	相関 ID（本番サービス用の OTel トレース ID）
message	string	人間が読める説明
context	object	操作固有のメタデータ
metrics	object	定量データ（カウント、期間）
error	object/null	失敗した場合の例外詳細

関連リソース

• Health Endpoint Monitoring Pattern - Microsoft Azure Architecture Center
• OWASP Logging Cheat Sheet - セキュリティベストプラクティス
• Write-Ahead Log pattern - Martin Fowler
• DuckDB JSON support - JSONL での SQL 分析
• lnav - SQL を備えたターミナルログファイルナビゲータ
• llm CLI - ログを LLM にパイプして分析
• structlog docs - 本番サービス向けの構造化ログ
• Pydantic Logfire - OTel ベースの AI/LLM オブザーバビリティ
• Langfuse - オープンソース LLM オブザーバビリティ（自社ホスト可能）

避けるべきアンチパターン

1. 無限のログ - 常にローテーション（ローカル）または stdout（コンテナ）を設定
2. 完全なシークレットをログに記録 - トークンフィンガープリントを使用。正規表現リダクションはバックストップ、主要ではない
3. < 5 個の低ボリュームサービスに loguru/structlog を追加 - print + JSONL で十分。依存関係は無料ではない
4. シークレットなしの裸の例外処理 - 特定の例外をキャッチしてログに記録
5. サイレント失敗 - エラーを抑制する前にログに記録
6. enqueue=True で logger.complete() なし - サイレントログ損失がシャットダウンで発生
7. 遅いシンクで enqueue=True - 無限メモリ増殖
8. > 10k イベント/秒で json.dumps() - 2～10 倍の高速化で orjson を使用
9. OTel サービスで UUID4 トレース ID - OTel によって伝播されたトレース ID を使用
10. < 5 サービス用の Prometheus/Grafana - ヘルスエンドポイント + Uptime Kuma で十分
11. WAL とテレメトリを混同 - WAL はクラッシュ回復用（一時的）、テレメトリは監査用（永続的）

---

トラブルシューティング

問題	原因	解決方法
loguru が見つからない	インストールされていない	uv add loguru を実行
ログが表示されていない	ログレベルが間違っている	トラブルシューティング用に DEBUG レベルを設定
ログローテーションが機能しない	ローテーション設定が不足	logger.add() にrotation パラメータを追加
JSONL 解析エラー	不正なログ行	エスケープされていない特殊文字を確認
enqueue=True での OOM	無限内部キュー	RSS を監視；structlog を使用または遅いシンクを回避
シャットダウンで ログ損失	logger.complete() が不足	await logger.complete() または logger.remove() を呼び出し
JSONL シリアライゼーションが遅い	高ボリュームで stdlib json	orjson.dumps().decode() に切り替え
ログにシークレットが存在	フィンガープリントなし	トークン全体ではなくスライスをログに記録
journald がキャプチャしない	フラッシュが不足	print(..., flush=True) または PYTHONUNBUFFERED=1 を使用
サービスクラッシュ時にアラートなし	外部モニタ不足	Uptime Kuma または Gatus をポーリングヘルスエンドポイント用に追加

実行後の反省

このスキルが完了した後、閉じる前に確認してください:

1. コマンドは成功しましたか？ — そうでない場合は、エラーを引き起こした指示またはエラーテーブルを修正します。
2. パラメータまたは出力が変わりましたか？ — 基盤となるツールのインターフェースが変動した場合は、使用例とパラメータテーブルを更新して一致させます。
3. 回避策が必要でしたか？ — 即興が必要だった場合（異なるフラグ、追加ステップ）、次の呼び出しで同じ回避策が不要になるように SKILL.md を更新します。

実際に再現可能な問題の場合のみ更新してください — 投機的ではなく。