Patrón de Caché de Archivos basado en Hash de Contenido
Almacena en caché resultados costosos de procesamiento de archivos (análisis de PDF, extracción de texto, análisis de imágenes) utilizando hashes SHA-256 del contenido como claves de caché. A diferencia del almacenamiento en caché basado en la ruta, este enfoque sobrevive a cambios de nombre o movimientos de archivos e invalida automáticamente cuando el contenido cambia.
Cuándo activarlo
- Construir pipelines de procesamiento de archivos (PDF, imágenes, extracción de texto)
- El costo de procesamiento es alto y los mismos archivos se procesan repetidamente
- Necesitas una opción
--cache/--no-cacheen la CLI - Quieres agregar almacenamiento en caché a funciones puras existentes sin modificarlas
Patrón Principal
1. Clave de Caché basada en Hash de Contenido
Usa el contenido del archivo (no la ruta) como clave de caché:
import hashlib
from pathlib import Path
_HASH_CHUNK_SIZE = 65536 # Fragmentos de 64KB para archivos grandes
def compute_file_hash(path: Path) -> str:
"""SHA-256 del contenido del archivo (dividido en fragmentos para archivos grandes)."""
if not path.is_file():
raise FileNotFoundError(f"Archivo no encontrado: {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()
¿Por qué hash de contenido? Renombrar/mover archivo = acierto de caché. Cambio de contenido = invalidación automática. No se necesita archivo de índice.
2. Dataclass Congelada para Entrada de Caché
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class CacheEntry:
file_hash: str
source_path: str
document: ExtractedDocument # El resultado almacenado en caché
3. Almacenamiento de Caché basado en Archivos
Cada entrada de caché se almacena como {hash}.json — búsqueda O(1) por hash, sin necesidad de archivo de índice.
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 # Tratar corrupción como fallo de caché
4. Envoltorio de Capa de Servicio (SRP)
Mantén la función de procesamiento pura. Agrega el almacenamiento en caché como una capa de servicio separada.
def extract_with_cache(
file_path: Path,
*,
cache_enabled: bool = True,
cache_dir: Path = Path(".cache"),
) -> ExtractedDocument:
"""Capa de servicio: verificación de caché -> extracción -> escritura de caché."""
if not cache_enabled:
return extract_text(file_path) # Función pura, sin conocimiento de caché
file_hash = compute_file_hash(file_path)
# Verificar caché
cached = read_cache(cache_dir, file_hash)
if cached is not None:
logger.info("Acierto de caché: %s (hash=%s)", file_path.name, file_hash[:12])
return cached.document
# Fallo de caché -> extraer -> almacenar
logger.info("Fallo de caché: %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
Decisiones Clave de Diseño
| Decisión | Razón |
|---|---|
| Hash de contenido SHA-256 | Independiente de la ruta, se invalida automáticamente al cambiar el contenido |
Nombrado de archivos {hash}.json | Búsqueda O(1), no necesita archivo de índice |
| Envoltorio de capa de servicio | SRP: la extracción permanece pura, la caché es una preocupación separada |
| Serialización JSON manual | Control total sobre la serialización de dataclass congelada |
La corrupción retorna None | Degradación elegante, se reprocesa en la siguiente ejecución |
cache_dir.mkdir(parents=True) | Creación perezosa del directorio en la primera escritura |
Mejores Prácticas
- Hashea el contenido, no las rutas — las rutas cambian, la identidad del contenido no
- Divide archivos grandes en fragmentos al hashear — evita cargar archivos enteros en memoria
- Mantén las funciones de procesamiento puras — no deben saber nada sobre el almacenamiento en caché
- Registra aciertos/fallos de caché con hashes truncados para depuración
- Maneja la corrupción con elegancia — trata las entradas de caché inválidas como fallos, nunca falles
Anti-patrones a Evitar
# MALO: Caché basada en ruta (falla al mover/renombrar archivo)
cache = {"/ruta/al/archivo.pdf": resultado}
# MALO: Agregar lógica de caché dentro de la función de procesamiento (violación SRP)
def extract_text(path, *, cache_enabled=False, cache_dir=None):
if cache_enabled: # Ahora esta función tiene dos responsabilidades
...
# MALO: Usar dataclasses.asdict() con dataclasses congeladas anidadas
# (puede causar problemas con tipos anidados complejos)
data = dataclasses.asdict(entry) # Usa serialización manual en su lugar
Cuándo Usar
- Pipelines de procesamiento de archivos (análisis de PDF, OCR, extracción de texto, análisis de imágenes)
- Herramientas CLI que se beneficien de opciones
--cache/--no-cache - Procesamiento por lotes donde los mismos archivos aparecen en múltiples ejecuciones
- Agregar almacenamiento en caché a funciones puras existentes sin modificarlas
Cuándo NO Usar
- Datos que siempre deben estar frescos (fuentes en tiempo real)
- Entradas de caché que serían extremadamente grandes (considera la transmisión en su lugar)
- Resultados que dependen de parámetros más allá del contenido del archivo (por ejemplo, diferentes configuraciones de extracción)