content-hash-cache-pattern

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Content-Hash File Cache Pattern

内容哈希文件缓存模式

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

—

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

提取日期: 2026-02-10 上下文: 基于SHA-256哈希缓存文件处理结果，并通过服务层进行包装的模式

Problem / 課題

问题

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

python

undefined

文件处理（PDF解析、文本提取等）耗时较长，但重复处理同一文件属于无效操作：

python

undefined

WRONG: 毎回フルパイプライン実行

错误：每次都执行完整流水线

def process_file(path: Path) -> Result: return expensive_extraction(path) # Always re-runs

def process_file(path: Path) -> Result: return expensive_extraction(path) # 始终重复执行

WRONG: パスベースキャッシュ（ファイル移動で無効化）

错误：基于路径的缓存（文件移动后失效）

cache = {"/path/to/file.pdf": result} # Path changes → cache miss

cache = {"/path/to/file.pdf": result} # 路径变更 → 缓存未命中

WRONG: 既存関数にキャッシュパラメータ追加（SRP違反）

错误：为现有函数添加缓存参数（违反单一职责原则）

def extract_text(path, *, cache_enabled=False, cache_dir=None): if cache_enabled: # Extraction function now has cache responsibility ...

---

def extract_text(path, *, cache_enabled=False, cache_dir=None): if cache_enabled: # 提取函数现在承担了缓存职责 ...

---

Solution / 解決策

解决方案

1. Content-Hash Based Cache Key

1. 基于内容哈希的缓存键

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

python

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()

利点: ファイル移動・リネームでもキャッシュヒット、内容変更で自動無効化

不使用文件路径，而是将文件内容的SHA-256哈希作为缓存键：

python

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

2. 用于缓存条目的冻结数据类

python

from dataclasses import dataclass

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

python

from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class CacheEntry:
    file_hash: str
    source_path: str
    document: ExtractedDocument  # 缓存的结果

3. JSON Serialization of Frozen Dataclasses

3. 冻结数据类的JSON序列化

dataclasses.asdict()

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

python

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,
    )

dataclasses.asdict()

在处理嵌套的冻结数据类时会出现问题，因此采用手动映射：

python

import json
from typing import Any

def _serialize_entry(entry: CacheEntry) -> dict[str, Any]:
    """手动映射以完全控制序列化格式。"""
    doc = entry.document
    return {
        "file_hash": entry.file_hash,
        "source_path": entry.source_path,
        "document": {
            "text": doc.text,
            "chunks": list(doc.chunks),  # 元组转列表以支持JSON
            "file_type": doc.file_type,
            # ... 其他字段
        },
    }

def _deserialize_entry(data: dict[str, Any]) -> CacheEntry:
    doc_data = data["document"]
    document = ExtractedDocument(
        text=doc_data["text"],
        chunks=tuple(doc_data["chunks"]),  # 列表转元组
        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)

4. 服务层包装器（遵循单一职责原则）

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

python

undefined

不修改纯处理函数，在服务层包装缓存逻辑：

python

undefined

service.py — cache wrapper

service.py — 缓存包装器

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

undefined

def extract_with_cache(file_path: Path, *, config: AppConfig) -> ExtractedDocument: """服务层：缓存检查 → 提取 → 写入缓存。""" if not config.cache_enabled: return extract_text(file_path) # 纯函数，无需知晓缓存逻辑

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

undefined

5. Graceful Corruption Handling

5. 优雅的损坏处理

python

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

python

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("缓存条目损坏: %s", cache_file)
        return None  # 将损坏视为缓存未命中

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

设计要点

Choice / 選択	Reason / 理由
SHA-256 content hash	Path-independent, auto-invalidates on content change
`{hash}.json` file naming	O(1) lookup, no index file needed
Service layer wrapper	SRP: extraction stays pure, cache is separate concern
Manual JSON serialization	Full control over frozen dataclass serialization
Corruption → None	Graceful degradation, re-extracts on next run
`cache_dir.mkdir(parents=True)`	Lazy directory creation on first write

选择	理由
SHA-256内容哈希	与路径无关，内容变更时自动失效
`{hash}.json` 文件命名	O(1)查找，无需索引文件
服务层包装器	遵循单一职责原则：提取函数保持纯净，缓存为独立关注点
手动JSON序列化	完全控制冻结数据类的序列化
损坏时返回None	优雅降级，下次运行时重新提取
`cache_dir.mkdir(parents=True)`	首次写入时自动创建目录

When to Use / 使用すべき場面

适用场景

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

文件处理流水线（PDF解析、图像处理、文本提取）
处理成本高，同一文件重复处理频繁的场景
CLI工具需要
```
--cache/--no-cache
```
选项的场景
为现有纯函数添加缓存的场景（维持单一职责原则）

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

不适用场景

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

需要实时更新的数据（始终需要最新内容）
缓存条目非常大的场景（占用过多内存/磁盘）
处理结果依赖文件内容以外参数的场景（配置变更时需要失效缓存）

content-hash-cache-pattern

Original

Translation

Content-Hash File Cache Pattern

内容哈希文件缓存模式

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

Problem / 課題

问题

WRONG: 毎回フルパイプライン実行

错误：每次都执行完整流水线

WRONG: パスベースキャッシュ（ファイル移動で無効化）

错误：基于路径的缓存（文件移动后失效）

WRONG: 既存関数にキャッシュパラメータ追加（SRP違反）

错误：为现有函数添加缓存参数（违反单一职责原则）

Solution / 解決策

解决方案

1. Content-Hash Based Cache Key

1. 基于内容哈希的缓存键

2. Frozen Dataclass for Cache Entry

2. 用于缓存条目的冻结数据类

3. JSON Serialization of Frozen Dataclasses

3. 冻结数据类的JSON序列化

4. Service Layer Wrapper (SRP)

4. 服务层包装器（遵循单一职责原则）

service.py — cache wrapper

service.py — 缓存包装器

5. Graceful Corruption Handling

5. 优雅的损坏处理

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

设计要点

When to Use / 使用すべき場面

适用场景

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

不适用场景

Related Patterns / 関連パターン

相关模式