コンテンツハッシュファイルキャッシュパターン
SHA-256 コンテンツハッシュをキャッシュキーとして使用し、高コストなファイル処理結果（PDF解析、テキスト抽出、画像分析など）をキャッシュします。パスベースのキャッシュとは異なり、このアプローチではファイルの移動や名前変更後もキャッシュが有効であり、コンテンツが変更されると自動的に無効化されます。
アクティベーションのタイミング
- ファイル処理パイプラインの構築（PDF、画像、テキスト抽出）
- 処理コストが高く、同じファイルが繰り返し処理される場合
--cache/--no-cacheCLI オプションが必要な場合- 既存の純粋関数を修正せずにキャッシュを追加したい場合
コアパターン
1. コンテンツハッシュベースのキャッシュキー
ファイルパスではなく、ファイル内容をキャッシュキーとして使用します。
import hashlib
from pathlib import Path
_HASH_CHUNK_SIZE = 65536 # 大きなファイル用の64KBチャンク
def compute_file_hash(path: Path) -> str:
"""ファイル内容のSHA-256ハッシュ（大きなファイルはチャンク読み込み）。"""
if not path.is_file():
raise FileNotFoundError(f"ファイルが見つかりません: {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 # キャッシュされた結果
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 # 破損はキャッシュミスとして扱う
4. サービス層ラッパー（SRP）
処理関数は純粋なまま保ち、キャッシュは別のサービス層として追加します。
def extract_with_cache(
file_path: Path,
*,
cache_enabled: bool = True,
cache_dir: Path = Path(".cache"),
) -> ExtractedDocument:
"""サービス層: キャッシュチェック -> 抽出 -> キャッシュ書き込み。"""
if not cache_enabled:
return extract_text(file_path) # 純粋関数、キャッシュを知らない
file_hash = compute_file_hash(file_path)
# キャッシュをチェック
cached = read_cache(cache_dir, file_hash)
if cached is not None:
logger.info("キャッシュヒット: %s (hash=%s)", file_path.name, file_hash[:12])
return cached.document
# キャッシュミス -> 抽出 -> 保存
logger.info("キャッシュミス: %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) ルックアップ、インデックスファイル不要 |
| サービス層ラッパー | SRP: 抽出は純粋関数のまま、キャッシュは別の関心事 |
| 手動 JSON シリアライズ | 不変データクラスのシリアライズを完全制御 |
破損時は None を返す | グレースフルデグラデーション、次回実行時に再処理 |
cache_dir.mkdir(parents=True) | 初回書き込み時に遅延ディレクトリ作成 |
ベストプラクティス
- パスではなくコンテンツをハッシュ化 — パスは変わっても、コンテンツの同一性は変わらない
- 大きなファイルのハッシュ化はチャンク読み込み — ファイル全体をメモリに読み込まない
- 処理関数は純粋に保つ — キャッシュについて一切知る必要がない
- キャッシュヒット/ミスをログに記録 — デバッグのために短縮ハッシュを使用
- 破損をグレースフルに処理 — 無効なキャッシュエントリはミスとして扱い、クラッシュさせない
避けるべきアンチパターン
# 悪い例: パスベースのキャッシュ（ファイル移動・名前変更で壊れる）
cache = {"/path/to/file.pdf": result}
# 悪い例: 処理関数内部にキャッシュロジックを追加（SRP違反）
def extract_text(path, *, cache_enabled=False, cache_dir=None):
if cache_enabled: # この関数に2つの責務が生じる
...
# 悪い例: ネストした不変データクラスに dataclasses.asdict() を使用
# （複雑なネスト型で問題を引き起こす可能性がある）
data = dataclasses.asdict(entry) # 代わりに手動シリアライズを使用
使用すべき場合
- ファイル処理パイプライン（PDF解析、OCR、テキスト抽出、画像分析）
--cache/--no-cacheオプションが有用なCLIツール- 実行をまたいで同じファイルが現れるバッチ処理
- 既存の純粋関数を変更せずにキャッシュを追加する場合
使用すべきでない場合
- 常に最新データが必要な場合（リアルタイムフィード）
- キャッシュエントリが極端に大きい場合（ストリーミングを検討）
- ファイル内容以外のパラメータに依存する結果の場合（例：異なる抽出設定）