padrão-de-cache-por-hash-de-conteúdo

Padrão de Cache de Arquivos por Hash de Conteúdo

Armazene em cache resultados caros de processamento de arquivos (análise de PDF, extração de texto, análise de imagens) usando hashes SHA-256 do conteúdo como chaves de cache. Ao contrário do cache baseado em caminho, esta abordagem sobrevive a movimentações/renomeações de arquivos e invalida automaticamente quando o conteúdo muda.

Quando Ativar

• Construindo pipelines de processamento de arquivos (PDF, imagens, extração de texto)
• Custo de processamento é alto e os mesmos arquivos são processados repetidamente
• Necessidade de uma opção --cache/--no-cache na CLI
• Deseja adicionar cache a funções puras existentes sem modificá-las

Padrão Principal

1. Chave de Cache Baseada em Hash de Conteúdo

Use o conteúdo do arquivo (não o caminho) como chave de cache:

import hashlib
from pathlib import Path

_HASH_CHUNK_SIZE = 65536  # chunks de 64 KB para arquivos grandes

def compute_file_hash(path: Path) -> str:
    """SHA-256 do conteúdo do arquivo (em chunks para arquivos grandes)."""
    if not path.is_file():
        raise FileNotFoundError(f"Arquivo não 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 que hash de conteúdo? Renomear/mover arquivo = acerto no cache. Mudança de conteúdo = invalidação automática. Nenhum arquivo de índice necessário.

2. Dataclass Imutável para Entrada de Cache

from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class CacheEntry:
    file_hash: str
    source_path: str
    document: ExtractedDocument  # O resultado em cache

3. Armazenamento de Cache Baseado em Arquivo

Cada entrada de cache é armazenada como {hash}.json — consulta O(1) por hash, nenhum arquivo de índice necessário.

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 corrupção como erro de cache

4. Invólucro de Camada de Serviço (SRP)

Mantenha a função de processamento pura. Adicione o cache como uma camada de serviço separada.

def extract_with_cache(
    file_path: Path,
    *,
    cache_enabled: bool = True,
    cache_dir: Path = Path(".cache"),
) -> ExtractedDocument:
    """Camada de serviço: verificação de cache -> extração -> escrita no cache."""
    if not cache_enabled:
        return extract_text(file_path)  # Função pura, sem conhecimento de cache

    file_hash = compute_file_hash(file_path)

    # Verificar cache
    cached = read_cache(cache_dir, file_hash)
    if cached is not None:
        logger.info("Acerto no cache: %s (hash=%s)", file_path.name, file_hash[:12])
        return cached.document

    # Erro no cache -> extrair -> armazenar
    logger.info("Erro no cache: %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

Decisões de Design Principais

Decisão	Justificativa
Hash de conteúdo SHA-256	Independente de caminho, invalida automaticamente na mudança de conteúdo
Nomeação de arquivo {hash}.json	Consulta O(1), nenhum arquivo de índice necessário
Invólucro de camada de serviço	SRP: extração permanece pura, cache é uma preocupação separada
Serialização JSON manual	Controle total sobre a serialização da dataclass imutável
Corrupção retorna None	Degradação graciosa, reprocessa na próxima execução
cache_dir.mkdir(parents=True)	Criação preguiçosa do diretório na primeira escrita

Melhores Práticas

• Aplique hash no conteúdo, não nos caminhos — caminhos mudam, a identidade do conteúdo não
• Divida arquivos grandes em chunks ao aplicar hash — evite carregar arquivos inteiros na memória
• Mantenha funções de processamento puras — elas não devem saber nada sobre cache
• Registre acerto/erro de cache com hashes truncados para depuração
• Lide com corrupção graciosamente — trate entradas de cache inválidas como erros, nunca cause falhas

Anti-Padrões a Evitar

# RUIM: Cache baseado em caminho (quebra ao mover/renomear arquivo)
cache = {"/caminho/para/arquivo.pdf": resultado}

# RUIM: Adicionar lógica de cache dentro da função de processamento (violação SRP)
def extract_text(path, *, cache_enabled=False, cache_dir=None):
    if cache_enabled:  # Agora esta função tem duas responsabilidades
        ...

# RUIM: Usar dataclasses.asdict() com dataclasses imutáveis aninhadas
# (pode causar problemas com tipos aninhados complexos)
data = dataclasses.asdict(entry)  # Use serialização manual

Quando Usar

• Pipelines de processamento de arquivos (análise de PDF, OCR, extração de texto, análise de imagens)
• Ferramentas de linha de comando que se beneficiam de opções --cache/--no-cache
• Processamento em lote onde os mesmos arquivos aparecem em execuções diferentes
• Adicionar cache a funções puras existentes sem modificá-las

Quando NÃO Usar

• Dados que devem estar sempre atualizados (feeds em tempo real)
• Entradas de cache que seriam extremamente grandes (considere streaming)
• Resultados que dependem de parâmetros além do conteúdo do arquivo (por exemplo, configurações de extração diferentes)