Content-Hash-Cache-Pattern

Content-Hash-Datei-Cache-Muster

Zwischenspeichern Sie teure Dateiverarbeitungsergebnisse (PDF-Parsing, Textextraktion, Bildanalyse) mit SHA-256-Inhalts-Hashes als Cache-Schlüssel. Im Gegensatz zur pfadbasierten Zwischenspeicherung überlebt dieser Ansatz Dateiverschiebungen/-umbenennungen und macht den Cache automatisch ungültig, wenn sich der Inhalt ändert.

Wann zu aktivieren

• Erstellung von Dateiverarbeitungs-Pipelines (PDF, Bilder, Textextraktion)
• Verarbeitungskosten sind hoch und dieselben Dateien werden wiederholt verarbeitet
• Bedarf an einer --cache/--no-cache CLI-Option
• Hinzufügen von Caching zu bestehenden reinen Funktionen ohne diese zu ändern

Kernmuster

1. Content-Hash-basierter Cache-Schlüssel

Verwenden Sie den Dateiinhalt (nicht den Pfad) als Cache-Schlüssel:

import hashlib
from pathlib import Path

_HASH_CHUNK_SIZE = 65536  # 64KB Blöcke für große Dateien

def compute_file_hash(path: Path) -> str:
    """SHA-256 des Dateiinhalts (blockweise für große Dateien)."""
    if not path.is_file():
        raise FileNotFoundError(f"Datei nicht gefunden: {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()

Warum Content-Hash? Dateiumbenennung/-verschiebung = Cache-Treffer. Inhaltsänderung = automatische Ungültigkeit. Keine Indexdatei erforderlich.

2. Frozen-Dataclass für Cache-Eintrag

from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class CacheEntry:
    file_hash: str
    source_path: str
    document: ExtractedDocument  # Das zwischengespeicherte Ergebnis

3. Dateibasierte Cache-Speicherung

Jeder Cache-Eintrag wird als {hash}.json gespeichert — O(1)-Lookup nach Hash, keine Indexdatei erforderlich.

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  # Korruption als Cache-Fehltreffer behandeln

4. Service-Schicht-Wrapper (SRP)

Halten Sie die Verarbeitungsfunktion rein. Fügen Sie das Caching als separate Service-Schicht hinzu.

def extract_with_cache(
    file_path: Path,
    *,
    cache_enabled: bool = True,
    cache_dir: Path = Path(".cache"),
) -> ExtractedDocument:
    """Service-Schicht: Cache-Prüfung -> Extraktion -> Cache-Schreiben."""
    if not cache_enabled:
        return extract_text(file_path)  # Reine Funktion, kein Cache-Wissen

    file_hash = compute_file_hash(file_path)

    # Cache prüfen
    cached = read_cache(cache_dir, file_hash)
    if cached is not None:
        logger.info("Cache-Treffer: %s (hash=%s)", file_path.name, file_hash[:12])
        return cached.document

    # Cache-Fehltreffer -> extrahieren -> speichern
    logger.info("Cache-Fehltreffer: %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

Wichtige Designentscheidungen

Entscheidung	Begründung
SHA-256 Content-Hash	Pfadunabhängig, automatische Ungültigkeit bei Inhaltsänderung
{hash}.json Dateibenennung	O(1)-Lookup, keine Indexdatei erforderlich
Service-Schicht-Wrapper	SRP: Extraktion bleibt rein, Cache ist separate Sache
Manuelle JSON-Serialisierung	Volle Kontrolle über Serialisierung von Frozen-Dataclasses
Korruption gibt None zurück	Sanfte Degradierung, Neuverarbeitung beim nächsten Lauf
cache_dir.mkdir(parents=True)	Verzögerte Verzeichniserstellung beim ersten Schreibvorgang

Best Practices

• Hashen Sie Inhalte, nicht Pfade — Pfade ändern sich, Inhaltsidentität nicht
• Verarbeiten Sie große Dateien blockweise beim Hashing — vermeiden Sie das Laden ganzer Dateien in den Speicher
• Halten Sie Verarbeitungsfunktionen rein — sie sollten nichts über das Caching wissen
• Protokollieren Sie Cache-Treffer/-Fehltreffer mit verkürzten Hashes zum Debuggen
• Behandeln Sie Korruption elegant — behandeln Sie ungültige Cache-Einträge als Fehltreffer, nie als Absturz

Anti-Patterns zu vermeiden

# SCHLECHT: Pfadbasiertes Caching (bricht bei Dateiverschiebung/-umbenennung)
cache = {"/pfad/zu/datei.pdf": ergebnis}

# SCHLECHT: Cache-Logik in die Verarbeitungsfunktion einbauen (SRP-Verletzung)
def extract_text(pfad, *, cache_aktiviert=False, cache_verzeichnis=None):
    if cache_aktiviert:  # Jetzt hat diese Funktion zwei Verantwortlichkeiten
        ...

# SCHLECHT: Verwendung von dataclasses.asdict() mit verschachtelten Frozen-Dataclasses
# (kann bei komplexen verschachtelten Typen Probleme verursachen)
daten = dataclasses.asdict(eintrag)  # Stattdessen manuelle Serialisierung verwenden

Wann zu verwenden

• Dateiverarbeitungs-Pipelines (PDF-Parsing, OCR, Textextraktion, Bildanalyse)
• CLI-Tools, die von Optionen wie --cache/--no-cache profitieren
• Stapelverarbeitung, bei der dieselben Dateien über mehrere Läufe hinweg erscheinen
• Hinzufügen von Caching zu bestehenden reinen Funktionen ohne diese zu ändern

Wann NICHT zu verwenden

• Daten, die immer aktuell sein müssen (Echtzeit-Feeds)
• Cache-Einträge, die extrem groß wären (alternativ Streaming in Betracht ziehen)
• Ergebnisse, die von Parametern über den Dateiinhalt hinaus abhängen (z. B. unterschiedliche Extraktionskonfigurationen)