python-logging-best-practices

Python ロギングのベストプラクティス

自己進化型スキル：このスキルは使用するたびに改善されます。手順に誤りがある場合、パラメーターが変化した場合、または回避策が必要になった場合は、先送りせず、すぐにこのファイルを修正してください。実際に再現できる問題がある場合にのみ更新してください。

このスキルを使う場面

以下の場合に、このスキルを使ってください。

• サービスやスクリプトに Python のロギングを設定するとき
• 分析用の構造化 JSONL ロギングを設定するとき
• ログローテーションを実装するとき
• 軽量（依存関係なし）と高機能なロギングのどちらを選ぶか判断するとき
• コンテナー化されたアプリケーション、systemd アプリケーション、またはローカルアプリケーションにロギングを追加するとき

概要

機械可読性（Claude Code による分析）と運用上の信頼性に最適化された、Python ロギングパターンの統一リファレンスです。実用可能な最小構成から始め、必要な場合にのみ拡張します。

判断のヒューリスティック：軽量な構成から始めて拡張する

Is it < 5 services on a single machine, < 1 event/sec?
  YES → Lightweight Pattern (print + JSONL telemetry)
  NO  → Is it containerized / serverless?
    YES → stdout JSON (any library), no file rotation
    NO  → Is OTel tracing required?
      YES → structlog + OTel
      NO  → loguru (CLI tools) or stdlib RotatingFileHandler

アプローチ	用途	メリット	デメリット
軽量	小規模な systemd サービス、セルフホスト、単独の運用担当者	依存関係なし、journald との統合、最小限のコード	重大度によるフィルタリングなし、モジュールごとの制御なし
loguru	CLI ツール、スクリプト、ローカルサービス	設定不要、ローテーション機能を内蔵、優れた開発者体験	外部依存、スキーマを厳密には強制しない
structlog	本番サービス、OTel 統合	ContextVars、プロセッサーチェーン、OTel ネイティブ	学習コストが高い
stdlib	LaunchAgent デーモン、依存関係を追加できない制約	依存関係なし、Python 3.14 の merge_extra	ボイラープレートが多い、構造化のデフォルト設定なし
Logfire	AI／LLM のオブザーバビリティ、Pydantic アプリ	OTel を基盤とし、トークン／コスト追跡と SQL に対応	SaaS への依存、新しいエコシステム

---

推奨：軽量パターン（依存関係なし）

対象：systemd サービスが 5 未満、サーバー 1 台、運用担当者 1 名。relay-monitor の本番環境で十分に検証されています。

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

• チャネル 1：print(flush=True) → systemd journald（運用ログ、人間が読みやすい形式）
• チャネル 2：追記専用の JSONL ファイル（構造化テレメトリー、機械可読形式）

これは、12-Factor App の「ログをイベントストリームとして扱う」という原則に対応しています。journald は運用（ローテーション、フィルタリング、メタデータ）を担い、JSONL ファイルは事後分析用のドメインテレメトリーを提供します。

アーキテクチャ：3 つの関心事を分離

関心事	仕組み	目的	ライフサイクル
運用ログ	print() → journald	人によるデバッグ、journalctl -u service -f	journald が管理（自動ローテーション）
テレメトリー	JSONL ファイル（telemetry.jsonl）	構造化された監査証跡、AI／LLM 分析	追記専用、サイズに応じてローテーション
状態復旧	WAL ファイル（任意）	不可逆操作のクラッシュ復旧	一時的。成功時に削除

軽量パターンの完全な例

"""Append-only JSONL telemetry logger with size-based rotation.

Zero external dependencies. Works with systemd journald for ops logging
and a separate JSONL file for structured machine-readable telemetry.
"""

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             # Keep 3 rotated backups (~30MB total)


def log_event(event_type: str, data: dict) -> None:
    """Append a structured JSON line to 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:
        # Fallback to stderr (captured by journald)
        print(f"[telemetry] write failed: {e}", file=__import__("sys").stderr, flush=True)


def _rotate() -> None:
    """Rotate telemetry files: .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 (goes to journald via stdout) ===

def log(msg: str) -> None:
    """Human-readable operational log line. Captured by journald."""
    ts = datetime.now(timezone.utc).strftime("%H:%M:%S")
    print(f"[{ts}] {msg}", flush=True)

使用方法：

# Operational (human reads via journalctl -u myservice -f)
log("Refreshing token for account X")
log("Switch: account A → account B (reason: 5h breach)")

# Telemetry (machine reads via 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:
    """Extract uniquely identifiable chars from a token's mid-section.

    The prefix (sk-ant-oat01-) and suffix (...AA) are common across tokens.
    Chars 14-22 (after the prefix) are the most unique per-token.
    Middle-slice avoids leaking type-prefix metadata that prefix-based
    approaches expose.
    """
    if len(token) > 25:
        return token[14:22]
    return token[:8] if token else ""

# Usage: log the fingerprint, never the token
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():
    """White-box monitoring — current state on demand."""
    return {"active_account": ..., "accounts": [...], "polled_at": ...}

@app.get("/api/vault-health")
def vault_health():
    """Token health for all accounts."""
    return {name: {"status": "healthy", "expires_in": "7.5h", ...} for ...}

これは Health Endpoint Monitoring Pattern（Microsoft Azure Architecture Center）／Health Check API Pattern（microservices.io）です。ダッシュボード自体が監視ツールであり、Grafana／Prometheus は不要です。

サービス自身が運用状態を構造化 JSON として提供すると、次のメリットがあります：

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

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

ログ集約スタックは不要です。以下の単一バイナリのツールは、JSONL を直接処理できます：

# DuckDB — SQL analytics on JSONL (most powerful)
duckdb -c "SELECT type, count(*) FROM read_json_auto('telemetry.jsonl') GROUP BY 1 ORDER BY 2 DESC"

# jq — ad-hoc JSON filtering
jq 'select(.type == "token_refresh")' telemetry.jsonl

# journalctl — already exports JSONL natively
journalctl -u ccmax-switcher -o json --since "1h ago" | jq 'select(.PRIORITY == "3")'

# lnav — interactive terminal log viewer with SQL
lnav telemetry.jsonl

# llm (Simon Willison) — pipe to LLM for AI post-mortem
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 パターン

ログライブラリを活用できる CLI ツール、スクリプト、サービス向け：

ログローテーション（ローカル／CLI アプリでは必ず設定）

from loguru import logger

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

# stdlib alternative (zero-dep)
from logging.handlers import RotatingFileHandler

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

コンテナー／サーバーレスアプリ：ファイルローテーションは完全に省略します。JSON 形式で stdout／stderr にログを出力し、収集とローテーションはコンテナーランタイムに任せます。

JSONL 形式（機械可読）

# One JSON object per line - jq-parseable
{"timestamp": "2026-01-14T12:45:23.456Z", "level": "info", "message": "..."}

ファイル拡張子：常に .jsonl を使用します（.json や .log ではありません）

パフォーマンス：毎秒 10,000 件を超えるレコードを処理する場合は、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())
# Sync: logger.remove()

Loguru + JSONL の完全な例

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.14"
# 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	失敗時の例外の詳細

関連リソース

• ヘルスエンドポイント監視パターン - Microsoft Azure Architecture Center
• OWASP ロギングチートシート - セキュリティのベストプラクティス
• 先行書き込みログパターン - Martin Fowler
• DuckDB の JSON サポート - JSONL の SQL 分析
• lnav - SQL に対応したターミナル用ログファイルナビゲーター
• llm CLI - ログを LLM にパイプして分析
• structlog ドキュメント - 本番サービス向け構造化ロギング
• Pydantic Logfire - OTel を基盤とする AI/LLM の可観測性
• Langfuse - オープンソースの LLM 可観測性（セルフホスト可能）

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

1. 上限のないログ - 必ずローテーション（ローカル）または stdout（コンテナー）を設定します
2. シークレット全体のロギング - トークンのフィンガープリントを使用します。正規表現によるマスキングは最後の手段であり、主な対策ではありません
3. トラフィックの少ない 5 未満のサービスに loguru/structlog を追加する - print + JSONL で十分です。依存関係の追加は無料ではありません
4. ログを記録しない裸の except - 特定の例外を捕捉して、ログに記録します
5. サイレントな失敗 - エラーを抑制する前にログに記録します
6. logger.complete() を使わない enqueue=True - シャットダウン時にログがひそかに失われます
7. 低速なシンクでの enqueue=True - メモリ使用量が際限なく増加します
8. イベント数が 1 秒あたり 10,000 件を超える場合の json.dumps() - orjson を使用すると 2～10 倍高速になります
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 のシリアライズが遅い	大量のデータに標準ライブラリの json を使用している	orjson.dumps().decode() に切り替える
ログにシークレットが含まれる	フィンガープリント処理がない	値全体ではなく、トークンの一部を記録する
journald が出力を取得しない	flush がない	print(..., flush=True) または PYTHONUNBUFFERED=1 を使用する
サービスがクラッシュしてもアラートがない	外部監視がない	Uptime Kuma または Gatus を追加して、ヘルスエンドポイントをポーリングする

実行後の振り返り

このスキルの完了後、終了する前に確認してください。

1. コマンドは成功しましたか？ — 成功しなかった場合は、失敗の原因となった指示またはエラーの表を修正します。
2. パラメーターや出力は変わりましたか？ — 基盤となるツールのインターフェースに変更があった場合は、使用例とパラメーターの表を合わせて更新します。
3. 回避策が必要でしたか？ — 即興で対応（別のフラグや追加手順など）した場合は、次回同じ回避策が不要になるよう、この SKILL.md を更新します。

問題が実際に発生し、再現可能な場合にのみ更新してください。推測で更新しないでください。