patrón-de-caché-por-hash-de-contenido

Patrón de Caché de Archivos por 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) usando hashes SHA-256 del contenido como claves de caché. A diferencia de la caché basada en ruta, este enfoque sobrevive a cambios de nombre/movimiento de archivos y se 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
• Se necesita una opción --cache/--no-cache en la CLI
• Se desea añadir 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 (fragmentado 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 índice.

2. Dataclass Inmutable para la Entrada de Caché

from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class CacheEntry:
    file_hash: str
    source_path: str
    document: ExtractedDocument  # El resultado 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 í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. Capa de Servicio Envolvente (SRP)

Mantén la función de procesamiento pura. Añade la 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: verificar caché -> extracción -> escribir caché."""
    if not cache_enabled:
        return extract_text(file_path)  # Función pura, sin conocimiento de la 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	Justificación
Hash SHA-256 del contenido	Independiente de la ruta, se invalida automáticamente al cambiar el contenido
Nombrado de archivos {hash}.json	Búsqueda O(1), sin necesidad de archivo índice
Capa de servicio envolvente	SRP: la extracción sigue siendo pura, la caché es una preocupación separada
Serialización manual a JSON	Control total sobre la serialización de dataclasses inmutables
Corrupción devuelve None	Degradación suave, 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
• Fragmenta archivos grandes al hashear — evita cargar archivos enteros en memoria
• Mantén las funciones de procesamiento puras — no deben saber nada sobre la caché
• Registra aciertos/fallos de caché con hashes truncados para depuración
• Maneja la corrupción con gracia — trata las entradas de caché inválidas como fallos, nunca falles

Anti-Patrones a Evitar

# MALO: Caché basada en ruta (se rompe al mover/renombrar archivos)
cache = {"/ruta/al/archivo.pdf": resultado}

# MALO: Añadir lógica de caché dentro de la función de procesamiento (violación de 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 inmutables anidadas
# (puede causar problemas con tipos anidados complejos)
data = dataclasses.asdict(entry)  # Usa serialización manual en su lugar

Cuándo Usarlo

• Pipelines de procesamiento de archivos (análisis de PDF, OCR, extracción de texto, análisis de imágenes)
• Herramientas CLI que se benefician de opciones --cache/--no-cache
• Procesamiento por lotes donde los mismos archivos aparecen en diferentes ejecuciones
• Añadir caché a funciones puras existentes sin modificarlas

Cuándo NO Usarlo

• Datos que siempre deben estar frescos (fuentes en tiempo real)
• Entradas de caché que serían extremadamente grandes (considera streaming en su lugar)
• Resultados que dependen de parámetros más allá del contenido del archivo (por ejemplo, diferentes configuraciones de extracción)