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

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

Armazene em cache resultados caros de processamento de arquivos (parsing de PDF, extração de texto, análise de imagem) usando hashes SHA-256 do conteúdo como chaves de cache. Diferente 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)
• O custo de processamento é alto e os mesmos arquivos são processados repetidamente
• Necessidade de uma opção CLI --cache/--no-cache
• Deseja adicionar cache a funções puras existentes sem modificá-las

Padrão Central

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

_TAMANHO_CHUNK_HASH = 65536  # 64KB chunks para arquivos grandes

def computar_hash_arquivo(caminho: Path) -> str:
    """SHA-256 do conteúdo do arquivo (em chunks para arquivos grandes)."""
    if not caminho.is_file():
        raise FileNotFoundError(f"Arquivo não encontrado: {caminho}")
    sha256 = hashlib.sha256()
    with open(caminho, "rb") as f:
        while True:
            chunk = f.read(_TAMANHO_CHUNK_HASH)
            if not chunk:
                break
            sha256.update(chunk)
    return sha256.hexdigest()

Por que hash de conteúdo? Renomear/mover arquivo = acerto de cache. Mudança de conteúdo = invalidação automática. Nenhum arquivo de índice necessário.

2. Dataclass Congelada para Entrada de Cache

from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class EntradaCache:
    hash_arquivo: str
    caminho_origem: str
    documento: DocumentoExtraido  # O resultado em cache

3. Armazenamento de Cache Baseado em Arquivo

Cada entrada de cache é armazenada como {hash}.json — consulta O(1) por hash, sem necessidade de arquivo de índice.

import json
from typing import Any

def escrever_cache(dir_cache: Path, entrada: EntradaCache) -> None:
    dir_cache.mkdir(parents=True, exist_ok=True)
    arquivo_cache = dir_cache / f"{entrada.hash_arquivo}.json"
    dados = serializar_entrada(entrada)
    arquivo_cache.write_text(json.dumps(dados, ensure_ascii=False), encoding="utf-8")

def ler_cache(dir_cache: Path, hash_arquivo: str) -> EntradaCache | None:
    arquivo_cache = dir_cache / f"{hash_arquivo}.json"
    if not arquivo_cache.is_file():
        return None
    try:
        raw = arquivo_cache.read_text(encoding="utf-8")
        dados = json.loads(raw)
        return desserializar_entrada(dados)
    except (json.JSONDecodeError, ValueError, KeyError):
        return None  # Trata corrupção como perda de cache

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

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

def extrair_com_cache(
    caminho_arquivo: Path,
    *,
    cache_ativado: bool = True,
    dir_cache: Path = Path(".cache"),
) -> DocumentoExtraido:
    """Camada de serviço: verificação de cache -> extração -> escrita no cache."""
    if not cache_ativado:
        return extrair_texto(caminho_arquivo)  # Função pura, sem conhecimento de cache

    hash_arquivo = computar_hash_arquivo(caminho_arquivo)

    # Verificar cache
    cacheado = ler_cache(dir_cache, hash_arquivo)
    if cacheado is not None:
        logger.info("Acerto de cache: %s (hash=%s)", caminho_arquivo.name, hash_arquivo[:12])
        return cacheado.documento

    # Perda de cache -> extrair -> armazenar
    logger.info("Perda de cache: %s (hash=%s)", caminho_arquivo.name, hash_arquivo[:12])
    doc = extrair_texto(caminho_arquivo)
    entrada = EntradaCache(hash_arquivo=hash_arquivo, caminho_origem=str(caminho_arquivo), documento=doc)
    escrever_cache(dir_cache, entrada)
    return doc

Decisões de Design Principais

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

Melhores Práticas

• Faça hash do conteúdo, não dos caminhos — caminhos mudam, a identidade do conteúdo não
• Processe arquivos grandes em chunks ao fazer hash — evite carregar arquivos inteiros na memória
• Mantenha funções de processamento puras — elas não devem saber nada sobre cache
• Registre acertos/perdas de cache com hashes truncados para depuração
• Lide com corrupção graciosamente — trate entradas de cache inválidas como perdas, nunca trave

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 extrair_texto(caminho, *, cache_ativado=False, dir_cache=None):
    if cache_ativado:  # Agora esta função tem duas responsabilidades
        ...

# RUIM: Usar dataclasses.asdict() com dataclasses congeladas aninhadas
# (pode causar problemas com tipos aninhados complexos)
dados = dataclasses.asdict(entrada)  # Use serialização manual em vez disso

Quando Usar

• Pipelines de processamento de arquivos (parsing de PDF, OCR, extração de texto, análise de imagem)
• Ferramentas CLI 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 frescos (alimentações 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 (ex.: diferentes configurações de extração)