Lógica de busca com cache em Python usando Redis

Entendendo o problema: por que cache importa em buscas

Implementando busca com cache básico

import redis
import json
from typing import List, Dict, Any

class BasicSearchCache:
    def __init__(self):
        self.redis = redis.Redis(
            host='localhost',
            port=6379,
            decode_responses=True
        )

    def buscar(self, query: str, limite: int = 10) -> List[Dict]:
        # Constrói chave de cache única
        chave_cache = f"busca:{query.lower()}:{limite}"

        # Tenta recuperar resultado armazenado
        resultado_cache = self.redis.get(chave_cache)
        if resultado_cache:
            return json.loads(resultado_cache)

        # Se não existe, consulta o banco de dados
        resultados = self._consultar_banco(query, limite)

        # Armazena por 1 hora
        self.redis.setex(
            chave_cache,
            3600,
            json.dumps(resultados)
        )

        return resultados

    def _consultar_banco(self, query: str, limite: int) -> List[Dict]:
        # Sua lógica de busca real aqui
        pass

Padrões avançados: cache com invalidação inteligente

Invalidação baseada em tempo (TTL)

class SearchWithTTL:
    def __init__(self):
        self.redis = redis.Redis(decode_responses=True)
        self.ttl_por_tipo = {
            'usuarios': 300,    # 5 minutos
            'produtos': 1800,   # 30 minutos
            'artigos': 3600     # 1 hora
        }

    def buscar(self, tipo: str, query: str) -> List[Dict]:
        chave = f"busca:{tipo}:{query.lower()}"

        resultado = self.redis.get(chave)
        if resultado:
            return json.loads(resultado)

        resultados = self._executar_busca(tipo, query)
        ttl = self.ttl_por_tipo.get(tipo, 600)

        self.redis.setex(chave, ttl, json.dumps(resultados))
        return resultados

    def _executar_busca(self, tipo: str, query: str) -> List[Dict]:
        pass

Invalidação reativa por eventos

class SearchWithEventInvalidation:
    def __init__(self):
        self.redis = redis.Redis(decode_responses=True)

    def buscar(self, query: str) -> List[Dict]:
        chave = f"busca:{query.lower()}"
        resultado = self.redis.get(chave)

        if resultado:
            return json.loads(resultado)

        resultados = self._executar_busca(query)
        self.redis.setex(chave, 3600, json.dumps(resultados))
        return resultados

    def atualizar_produto(self, produto_id: int, dados: Dict) -> None:
        # Atualiza o banco de dados
        self._salvar_banco(produto_id, dados)

        # Invalida os caches afetados pela mudança
        self._invalidar_caches_relacionados(produto_id)

    def _invalidar_caches_relacionados(self, produto_id: int) -> None:
        # Localiza todas as chaves potencialmente afetadas
        chaves_pattern = self.redis.keys("busca:*")

        for chave in chaves_pattern:
            # Aplica lógica para determinar se o cache é afetado
            self.redis.delete(chave)

Atenção: o uso de KEYS com padrão curinga em produção pode bloquear o Redis em bases grandes. Prefira SCAN para iteração segura e não bloqueante.

Otimizações avançadas: busca facetada com cache

class FacetedSearchCache:
    def __init__(self):
        self.redis = redis.Redis(decode_responses=True)

    def buscar(
        self,
        query: str,
        categoria: str = None,
        preco_min: int = None,
        preco_max: int = None,
        pagina: int = 1
    ) -> Dict[str, Any]:
        # Constrói chave determinística a partir dos filtros
        filtros = f"{categoria or 'all'}:{preco_min or 0}:{preco_max or 'inf'}"
        chave = f"busca:facetada:{query.lower()}:{filtros}:{pagina}"

        resultado = self.redis.get(chave)
        if resultado:
            return json.loads(resultado)

        dados = self._buscar_facetado(query, categoria, preco_min, preco_max, pagina)

        self.redis.setex(chave, 1800, json.dumps(dados))
        return dados

    def _buscar_facetado(
        self,
        query: str,
        categoria: str,
        preco_min: int,
        preco_max: int,
        pagina: int
    ) -> Dict[str, Any]:
        pass

Monitorando performance e cache hits

class MonitoredSearchCache:
    def __init__(self):
        self.redis = redis.Redis(decode_responses=True)
        self.stats_key = "cache:stats"

    def buscar(self, query: str) -> List[Dict]:
        chave = f"busca:{query.lower()}"

        # Incrementa o total de tentativas
        self.redis.incr(f"{self.stats_key}:tentativas")

        resultado = self.redis.get(chave)
        if resultado:
            self.redis.incr(f"{self.stats_key}:hits")
            return json.loads(resultado)

        # Registra miss e executa busca real
        self.redis.incr(f"{self.stats_key}:misses")

        resultados = self._executar_busca(query)
        self.redis.setex(chave, 3600, json.dumps(resultados))

        return resultados

    def obter_taxa_acerto(self) -> float:
        hits = int(self.redis.get(f"{self.stats_key}:hits") or 0)
        tentativas = int(self.redis.get(f"{self.stats_key}:tentativas") or 1)
        return (hits / tentativas) * 100 if tentativas > 0 else 0

    def _executar_busca(self, query: str) -> List[Dict]:
        pass

Melhores práticas e armadilhas a evitar

Cache que funciona de verdade

Próximos passos