modèle de cache par hash de contenu

Modèle de cache par hash de contenu de fichier

Mettez en cache les résultats coûteux du traitement de fichiers (analyse PDF, extraction de texte, analyse d'image) en utilisant des hachages SHA-256 du contenu comme clés de cache. Contrairement au cache basé sur le chemin, cette approche survit aux déplacements/renommages de fichiers et s'invalide automatiquement lorsque le contenu change.

Quand activer

• Construction de pipelines de traitement de fichiers (PDF, images, extraction de texte)
• Coût de traitement élevé et mêmes fichiers traités de manière répétée
• Besoin d'une option CLI --cache/--no-cache
• Ajout de cache à des fonctions pures existantes sans les modifier

Modèle de base

1. Clé de cache basée sur le hachage du contenu

Utilisez le contenu du fichier (pas le chemin) comme clé de cache :

import hashlib
from pathlib import Path

_HASH_CHUNK_SIZE = 65536  # Morceaux de 64 Ko pour les gros fichiers

def compute_file_hash(path: Path) -> str:
    """SHA-256 du contenu du fichier (en morceaux pour les gros fichiers)."""
    if not path.is_file():
        raise FileNotFoundError(f"Fichier introuvable : {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()

Pourquoi le hachage du contenu ? Renommage/déplacement du fichier = hit de cache. Changement de contenu = invalidation automatique. Aucun fichier d'index nécessaire.

2. Dataclass figée pour l'entrée de cache

from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class CacheEntry:
    file_hash: str
    source_path: str
    document: ExtractedDocument  # Le résultat mis en cache

3. Stockage du cache basé sur fichier

Chaque entrée de cache est stockée sous forme de {hash}.json — recherche en O(1) par hachage, aucun fichier d'index requis.

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  # Traite la corruption comme un miss de cache

4. Wrapper de couche de service (SRP)

Gardez la fonction de traitement pure. Ajoutez le cache comme une couche de service séparée.

def extract_with_cache(
    file_path: Path,
    *,
    cache_enabled: bool = True,
    cache_dir: Path = Path(".cache"),
) -> ExtractedDocument:
    """Couche de service : vérification du cache -> extraction -> écriture du cache."""
    if not cache_enabled:
        return extract_text(file_path)  # Fonction pure, sans connaissance du cache

    file_hash = compute_file_hash(file_path)

    # Vérifier le 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 -> extraction -> stockage
    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

Décisions de conception clés

Décision	Justification
Hachage SHA-256 du contenu	Indépendant du chemin, invalidation automatique lors du changement de contenu
Nommage de fichier {hash}.json	Recherche O(1), aucun fichier d'index nécessaire
Wrapper de couche de service	SRP : l'extraction reste pure, le cache est une préoccupation séparée
Sérialisation JSON manuelle	Contrôle total sur la sérialisation des dataclasses figées
La corruption retourne None	Dégradation progressive, re-traitement lors de l'exécution suivante
cache_dir.mkdir(parents=True)	Création paresseuse du répertoire lors de la première écriture

Bonnes pratiques

• Hachez le contenu, pas les chemins — les chemins changent, l'identité du contenu ne change pas
• Découpez les gros fichiers en morceaux lors du hachage — évitez de charger les fichiers entiers en mémoire
• Gardez les fonctions de traitement pures — elles ne doivent rien savoir du cache
• Enregistrez les hits/miss de cache avec des hachages tronqués pour le débogage
• Gérez la corruption avec élégance — traitez les entrées de cache invalides comme des misses, ne plantez jamais

Anti-modèles à éviter

# MAUVAIS : Cache basé sur le chemin (casse lors du déplacement/renommage du fichier)
cache = {"/chemin/vers/fichier.pdf": résultat}

# MAUVAIS : Ajout de logique de cache à l'intérieur de la fonction de traitement (violation SRP)
def extract_text(path, *, cache_enabled=False, cache_dir=None):
    if cache_enabled:  # Maintenant cette fonction a deux responsabilités
        ...

# MAUVAIS : Utilisation de dataclasses.asdict() avec des dataclasses figées imbriquées
# (peut causer des problèmes avec des types imbriqués complexes)
data = dataclasses.asdict(entry)  # Utilisez plutôt la sérialisation manuelle

Quand utiliser

• Pipelines de traitement de fichiers (analyse PDF, OCR, extraction de texte, analyse d'image)
• Outils CLI qui bénéficient d'options --cache/--no-cache
• Traitement par lots où les mêmes fichiers apparaissent d'une exécution à l'autre
• Ajout de cache à des fonctions pures existantes sans les modifier

Quand NE PAS utiliser

• Données qui doivent toujours être fraîches (flux en temps réel)
• Entrées de cache qui seraient extrêmement volumineuses (envisagez plutôt le streaming)
• Résultats qui dépendent de paramètres autres que le contenu du fichier (par exemple, différentes configurations d'extraction)