Content Hash Cache Pattern

Content-Hash File Cache Pattern

コンテンツハッシュキャッシュパターン

Extracted / 抽出日: 2026-02-10 Context / コンテキスト: ファイル処理結果をSHA-256ハッシュでキャッシュし、サービス層でラップするパターン

Problem / 課題

ファイル処理（PDF解析、テキスト抽出等）は時間がかかるが、同じファイルの再処理は無駄：

# WRONG: 毎回フルパイプライン実行
def process_file(path: Path) -> Result:
    return expensive_extraction(path)  # Always re-runs

# WRONG: パスベースキャッシュ（ファイル移動で無効化）
cache = {"/path/to/file.pdf": result}  # Path changes → cache miss

# WRONG: 既存関数にキャッシュパラメータ追加（SRP違反）
def extract_text(path, *, cache_enabled=False, cache_dir=None):
    if cache_enabled:  # Extraction function now has cache responsibility
        ...

Solution / 解決策

1. Content-Hash Based Cache Key

ファイルパスではなくファイル内容のSHA-256ハッシュをキーに使う：

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. Frozen Dataclass for Cache Entry

from dataclasses import dataclass

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

3. JSON Serialization of Frozen Dataclasses

dataclasses.asdict() はネストしたfrozen dataclassで問題が起きるため、手動マッピング：

import json
from typing import Any

def _serialize_entry(entry: CacheEntry) -> dict[str, Any]:
    """Manual mapping for full control over serialized format."""
    doc = entry.document
    return {
        "file_hash": entry.file_hash,
        "source_path": entry.source_path,
        "document": {
            "text": doc.text,
            "chunks": list(doc.chunks),  # tuple → list for JSON
            "file_type": doc.file_type,
            # ... other fields
        },
    }

def _deserialize_entry(data: dict[str, Any]) -> CacheEntry:
    doc_data = data["document"]
    document = ExtractedDocument(
        text=doc_data["text"],
        chunks=tuple(doc_data["chunks"]),  # list → tuple
        file_type=doc_data["file_type"],
    )
    return CacheEntry(
        file_hash=data["file_hash"],
        source_path=data["source_path"],
        document=document,
    )

4. Service Layer Wrapper (SRP)

純粋な処理関数を変更せず、サービス層でキャッシュロジックをラップ：

# service.py — cache wrapper
def extract_with_cache(file_path: Path, *, config: AppConfig) -> ExtractedDocument:
    """Service layer: cache check → extraction → cache write."""
    if not config.cache_enabled:
        return extract_text(file_path)  # Pure function, no cache knowledge

    cache_dir = Path(config.cache_dir)
    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

5. Graceful Corruption Handling

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):
        logger.warning("Corrupted cache entry: %s", cache_file)
        return None  # Treat corruption as cache miss

Key Design Choices / 設計上のポイント

When to Use / 使用すべき場面

• ファイル処理パイプライン（PDF解析、画像処理、テキスト抽出）
• 処理コストが高く、同一ファイルの再処理が頻繁な場合
• CLI ツールで --cache/--no-cache オプションが必要な場合
• 既存の純粋関数にキャッシュを追加する場合（SRP維持）

When NOT to Use / 使用すべきでない場面

• リアルタイム更新が必要なデータ（常に最新が必要）
• キャッシュエントリが非常に大きい場合（メモリ/ディスク圧迫）
• 処理結果がファイル内容以外のパラメータに依存する場合（設定変更でキャッシュ無効化が必要）

Related Patterns / 関連パターン

• python-immutable-accumulator.md — frozen dataclass + slotsパターン
• backward-compatible-frozen-extension.md — frozen dataclass拡張
• cost-aware-llm-pipeline.md — LLMパイプラインでのキャッシュ活用