内容哈希文件缓存模式
使用SHA-256内容哈希作为缓存键，缓存昂贵的文件处理结果（如PDF解析、文本提取、图像分析）。与基于路径的缓存不同，该方法在文件移动或重命名后依然有效，并在内容变化时自动失效。
何时激活
- 构建文件处理管道（PDF、图像、文本提取）
- 处理成本高，且相同文件被重复处理
- 需要一个
--cache/--no-cache命令行选项 - 希望在不修改现有纯函数的情况下为其添加缓存
核心模式
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. 服务层包装器（单一职责原则）
保持处理函数的纯粹性。将缓存作为独立的服务层添加。
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: # 现在这个函数有了两个职责
...
# 错误：对嵌套的不可变数据类使用 dataclasses.asdict()
# （可能导致复杂嵌套类型出现问题）
data = dataclasses.asdict(entry) # 应使用手动序列化
何时使用
- 文件处理管道（PDF解析、OCR、文本提取、图像分析）
- 受益于
--cache/--no-cache选项的命令行工具 - 跨多次运行处理相同文件的批处理场景
- 在不修改现有纯函数的情况下为其添加缓存
何时不使用
- 必须始终是最新数据的数据（实时数据流）
- 缓存条目极大（应考虑流式处理）
- 结果依赖于文件内容以外的参数（例如，不同的提取配置）