コンテンツハッシュファイルキャッシュパターン

SHA-256 コンテンツハッシュをキャッシュキーとして使用し、高コストなファイル処理結果（PDF 解析、テキスト抽出、画像分析）をキャッシュします。パスベースのキャッシュとは異なり、この方法はファイルの移動/名前変更後も有効であり、コンテンツ変更時には自動的に無効化されます。

いつ活用するか

• ファイル処理パイプラインを構築するとき（PDF、画像、テキスト抽出）
• 処理コストが高く、同じファイルが繰り返し処理されるとき
• --cache/--no-cache CLI オプションが必要なとき
• 既存の純粋関数を変更せずにキャッシュを追加したいとき

コアパターン

1. コンテンツハッシュベースのキャッシュキー

ファイルパスではなくファイルコンテンツをキャッシュキーとして使用します：

import hashlib
from pathlib import Path

_HASH_CHUNK_SIZE = 65536  # 64KB chunks for large files

def compute_file_hash(path: Path) -> str:
    """SHA-256 of file contents (chunked for large files)."""
    if not path.is_file():
        raise FileNotFoundError(f"File not found: {path}")
    sha256 = hashlib.sha256()
    with open(path, "rb") as f:
        while True:
            chunk = f.read(_HASH_CHUNK_SIZE)
            if not chunk:
                break
            sha256.update(chunk)
    return sha256.hexdigest()

コンテンツハッシュを使う理由は？ ファイルの名前変更/移動 = キャッシュヒット。コンテンツ変更 = 自動無効化。インデックスファイルが不要。

2. キャッシュエントリ用のフローズンデータクラス

from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class CacheEntry:
    file_hash: str
    source_path: str
    document: ExtractedDocument  # The cached result

3. ファイルベースのキャッシュストレージ

各キャッシュエントリは {hash}.json として保存——ハッシュで O(1) ルックアップを実現し、インデックスファイルは不要です。

import json
from typing import Any

def write_cache(cache_dir: Path, entry: CacheEntry) -> None:
    cache_dir.mkdir(parents=True, exist_ok=True)
    cache_file = cache_dir / f"{entry.file_hash}.json"
    data = serialize_entry(entry)
    cache_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")

def read_cache(cache_dir: Path, file_hash: str) -> CacheEntry | None:
    cache_file = cache_dir / f"{file_hash}.json"
    if not cache_file.is_file():
        return None
    try:
        raw = cache_file.read_text(encoding="utf-8")
        data = json.loads(raw)
        return deserialize_entry(data)
    except (json.JSONDecodeError, ValueError, KeyError):
        return None  # Treat corruption as cache miss

4. サービスレイヤーラッパー（単一責任の原則）

処理関数の純粋性を保ちます。キャッシュを独立したサービスレイヤーとして追加します。

def extract_with_cache(
    file_path: Path,
    *,
    cache_enabled: bool = True,
    cache_dir: Path = Path(".cache"),
) -> ExtractedDocument:
    """Service layer: cache check -> extraction -> cache write."""
    if not cache_enabled:
        return extract_text(file_path)  # Pure function, no cache knowledge

    file_hash = compute_file_hash(file_path)

    # Check cache
    cached = read_cache(cache_dir, file_hash)
    if cached is not None:
        logger.info("Cache hit: %s (hash=%s)", file_path.name, file_hash[:12])
        return cached.document

    # Cache miss -> extract -> store
    logger.info("Cache miss: %s (hash=%s)", file_path.name, file_hash[:12])
    doc = extract_text(file_path)
    entry = CacheEntry(file_hash=file_hash, source_path=str(file_path), document=doc)
    write_cache(cache_dir, entry)
    return doc

主要な設計決定

決定	理由
SHA-256 コンテンツハッシュ	パス非依存、コンテンツ変更時に自動無効化
{hash}.json ファイル命名	O(1) ルックアップ、インデックスファイル不要
サービスレイヤーラッパー	単一責任の原則：抽出機能は純粋、キャッシュは独立した関心事
手動 JSON シリアライズ	フローズンデータクラスのシリアライズを完全制御
破損時に None を返す	優雅に劣化、次回の実行時に再処理
cache_dir.mkdir(parents=True)	初回書き込み時にディレクトリを遅延作成

ベストプラクティス

• パスではなくコンテンツをハッシュ化する —— パスは変わるが、コンテンツ識別子は変わらない
• 大ファイルをハッシュする際はチャンク単位で処理 —— ファイル全体をメモリに読み込まない
• 処理関数の純粋性を保つ —— キャッシュについて何も知らないようにする
• キャッシュヒット/ミスをログに記録し、デバッグ用に切り詰めたハッシュ値を使用する
• 破損への対応は優雅に —— 無効なキャッシュエントリをミスとして扱い、決してクラッシュしない

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

# BAD: Path-based caching (breaks on file move/rename)
cache = {"/path/to/file.pdf": result}

# BAD: Adding cache logic inside the processing function (SRP violation)
def extract_text(path, *, cache_enabled=False, cache_dir=None):
    if cache_enabled:  # Now this function has two responsibilities
        ...

# BAD: Using dataclasses.asdict() with nested frozen dataclasses
# (can cause issues with complex nested types)
data = dataclasses.asdict(entry)  # Use manual serialization instead

適用シーン

• ファイル処理パイプライン（PDF 解析、OCR、テキスト抽出、画像分析）
• --cache/--no-cache オプションで利益を得る CLI ツール
• 複数回の実行で同じファイルが出現するバッチ処理
• 既存の純粋関数を変更せずにキャッシュを追加したい場合

非適用シーン

• 常に最新状態を保つ必要があるデータ（リアルタイムデータストリーム）
• キャッシュエントリが非常に大きくなる可能性がある場合（ストリーミング処理の検討が必要）
• 結果がファイルコンテンツ以外のパラメータに依存する場合（例：異なる抽出設定）