Redis com Python: Cache e Filas na Prática

Redis é o canivete suíço da infraestrutura: cache, filas, contadores, rate limit e sessões. Em projetos Python, ele resolve problemas de performance e escala com pouquíssimo código. Neste guia prático mostro como usar Redis de verdade, com exemplos que você aplica hoje.

Vou partir de um cenário comum: uma API que chama um modelo de IA caro e precisa responder rápido.

O que é Redis e por que ele é tão usado

• Em memória: leituras e escritas em microssegundos.
• Estruturas ricas: strings, listas, hash, sets, sorted sets.
• TTL nativo: dados que expiram sozinhos.
• Atômico: operações seguras para contadores e locks.

Subindo o Redis

docker run -d --name redis -p 6379:6379 redis:7-alpine
pip install redis

Conectando

import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)
r.ping()  # True

decode_responses=True devolve strings em vez de bytes — mais confortável.

Cache com TTL

O uso mais comum: guardar o resultado de algo caro por um tempo.

import json, hashlib

def chave(prompt: str) -> str:
    return "llm:" + hashlib.sha256(prompt.encode()).hexdigest()

def resposta_llm(prompt: str) -> str:
    k = chave(prompt)
    cache = r.get(k)
    if cache:
        return json.loads(cache)
    resultado = chamar_modelo(prompt)   # caro
    r.set(k, json.dumps(resultado), ex=3600)  # expira em 1h
    return resultado

Rate limit com INCR

Contar requisições por usuário e janela de tempo é trivial com INCR + EXPIRE.

def permitido(usuario: str, limite: int = 60) -> bool:
    k = f"rl:{usuario}:{int(time.time() // 60)}"
    atual = r.incr(k)
    if atual == 1:
        r.expire(k, 60)
    return atual <= limite

Filas simples com listas

Listas funcionam como fila FIFO para tarefas em background.

def enfileirar(tarefa: dict):
    r.lpush("fila:tarefas", json.dumps(tarefa))

def consumir(timeout: int = 5):
    item = r.brpop("fila:tarefas", timeout=timeout)
    return json.loads(item[1]) if item else None

Hash para objetos

r.hset("usuario:1", mapping={"nome": "Ana", "plano": "pro"})
print(r.hgetall("usuario:1"))

Sorted set para rankings

r.zadd("ranking", {"ana": 120, "bruno": 90, "carla": 150})
print(r.zrevrange("ranking", 0, 2, withscores=True))

Lock distribuído

Evitar que dois workers processem a mesma tarefa ao mesmo tempo:

def com_lock(nome: str, ttl: int = 30):
    return r.set(f"lock:{nome}", "1", nx=True, ex=ttl)

Cache de página/sessão

r.setex("sessao:abc", 1800, json.dumps({"user_id": 1}))
sessao = json.loads(r.get("sessao:abc") or "null")

Padrões de chave

Use prefixos consistentes: llm:, rl:, fila:, sessao:. Isso evita colisões e facilita limpeza.

Persistência e limites

Redis é memória — configure maxmemory e uma política de eviction (ex.: allkeys-lru). Para dados críticos, ative persistência (RDB/AOF) ou use outro armazenamento.

Erros comuns

• Usar Redis como banco principal de dados críticos.
• Esquecer TTL — cache cresce para sempre.
• Guardar objetos grandes (megabytes) — Redis é para dados pequenos e quentes.
• Não tratar indisponibilidade do Redis (degrade com fallback).
• Usar KEYS em produção — use SCAN.

Checklist

1. Defina padrão de chaves com prefixo.
2. Sempre pense no TTL.
3. Use operações atômicas para contadores.
4. Trate falha do Redis com fallback.
5. Configure maxmemory e eviction.

Pipeline completo: cache + fila + rate limit

Na prática, os três padrões se combinam. Veja uma API que recebe tarefas, limita por usuário, responde do cache quando possível e enfileira o resto.

from fastapi import FastAPI, HTTPException
app = FastAPI()

@app.post("/tarefa")
def criar(usuario: str, payload: dict):
    if not permitido(usuario):
        raise HTTPException(429, "Limite de requisicoes excedido")
    k = chave(str(payload))
    if cache := r.get(k):
        return {"origem": "cache", "resultado": json.loads(cache)}
    enfileirar({"usuario": usuario, "payload": payload})
    return {"origem": "fila", "status": "processando"}

Worker que consome a fila

def worker():
    while True:
        tarefa = consumir()
        if not tarefa:
            continue
        resultado = processar(tarefa["payload"])
        r.set(chave(str(tarefa["payload"])), json.dumps(resultado), ex=3600)

Padrão de fila confiável

Se o worker morrer no meio, a tarefa se perde. Para produção, use uma fila confiável (BRPOPLPUSH + lista de processamento) ou uma ferramenta dedicada como Celery/RQ.

def consumir_confiavel():
    item = r.brpoplpush("fila:tarefas", "fila:processando", timeout=5)
    return json.loads(item) if item else None

Contadores e métricas

r.incr("metricas:chamadas")
r.hincrby("metricas:dia", "2026-11-05", 1)

Pub/Sub para eventos

Além de fila, Redis faz broadcast de eventos em tempo real.

# publicar
r.publish("eventos", json.dumps({"tipo": "novo_pedido"}))

# assinar (em outro processo)
pubsub = r.pubsub()
pubsub.subscribe("eventos")
for msg in pubsub.listen():
    print(msg)

Observabilidade

Monitore INFO (memória, conexões, hit rate), latência das operações e taxa de eviction. Um hit rate baixo indica chaves mal escolhidas ou TTL curto demais.

Cache de sessão com FastAPI

from fastapi import FastAPI, Request
app = FastAPI()

@app.middleware("http")
async def sessao(request: Request, call_next):
    sid = request.cookies.get("sid")
    if sid:
        dados = r.get(f"sessao:{sid}")
        request.state.usuario = json.loads(dados) if dados else None
    return await call_next(request)

Rate limit por IP e por usuário

def permitido(chave_id: str, limite: int, janela: int) -> bool:
    k = f"rl:{chave_id}:{int(time.time() // janela)}"
    atual = r.incr(k)
    if atual == 1:
        r.expire(k, janela)
    return atual <= limite

Idempotência com SETNX

Evitar processar a mesma requisição duas vezes (ex.: pagamento).

def processar_uma_vez(txid: str) -> bool:
    return bool(r.set(f"tx:{txid}", "1", nx=True, ex=86400))

Leaderboard em tempo real

r.zincrby("placar", 10, "ana")
top = r.zrevrange("placar", 0, 9, withscores=True)

Redis com Docker Compose

services:
  api:
    build: .
    environment:
      REDIS_URL: redis://redis:6379/0
  redis:
    image: redis:7-alpine
    command: ["redis-server", "--maxmemory", "256mb", "--maxmemory-policy", "allkeys-lru"]

Fallback quando o Redis cai

def get_seguro(chave: str):
    try:
        return r.get(chave)
    except redis.RedisError:
        return None  # segue sem cache

Perguntas frequentes

Redis ou Memcached? Redis, pela variedade de estruturas e TTL.

Preciso de cluster? Só em escala grande; comece com uma instância e sentinela.

Perco dados no restart? Sim, sem persistência — por isso não é banco principal.

Cache de resultados de API externa

APIs de terceiros têm limite e custo. Cachear a resposta por alguns minutos reduz chamadas e melhora a latência.

def consultar_api(param: str, ttl: int = 300):
    k = f"api:{param}"
    if v := r.get(k):
        return json.loads(v)
    dados = requests.get(f"https://api.externa.com/{param}").json()
    r.set(k, json.dumps(dados), ex=ttl)
    return dados

Cache-aside vs write-through

Duas estratégias comuns:

• Cache-aside: a aplicação lê o cache e, no miss, busca na fonte e popula. Simples e flexível.
• Write-through: toda escrita atualiza cache e fonte. Consistência maior, mais complexo.

Invalidando cache corretamente

Cache velho é pior que cache nenhum. Padronize a invalidação:

def atualizar_produto(pid: int, dados: dict):
    salvar_no_banco(pid, dados)
    r.delete(f"produto:{pid}")   # invalida

Controlando concorrência com locks

Dois workers podem tentar a mesma tarefa ao mesmo tempo. Um lock distribuído evita processamento duplicado e condições de corrida.

import time, uuid

def adquirir_lock(nome: str, ttl: int = 30) -> str | None:
    token = uuid.uuid4().hex
    if r.set(f"lock:{nome}", token, nx=True, ex=ttl):
        return token
    return None

def liberar_lock(nome: str, token: str):
    # so libera se o token for o dono (evita liberar lock de outro)
    script = """if redis.call('get', KEYS[1]) == ARGV[1] then
                  return redis.call('del', KEYS[1]) else return 0 end"""
    r.eval(script, 1, f"lock:{nome}", token)

Esse padrão é essencial para jobs agendados que não podem rodar em duplicidade.

Cache de consultas pesadas ao banco

def relatorio_vendas(periodo: str):
    k = f"rel:{periodo}"
    if v := r.get(k):
        return json.loads(v)
    dados = consulta_lenta_no_banco(periodo)
    r.set(k, json.dumps(dados), ex=600)
    return dados

Estrutura de dados: qual usar quando

Necessidade	Estrutura
Valor simples	String
Objeto com campos	Hash
Fila FIFO	List
Itens únicos	Set
Ranking	Sorted Set
Contador com janela	String + EXPIRE

Streams e processamento em tempo real

Redis Streams permite processar eventos em ordem, com grupos de consumidores — ideal para pipelines.

def publicar_evento(tipo: str, dados: dict):
    r.xadd("eventos:stream", {"tipo": tipo, "dados": json.dumps(dados)})

def ler_eventos(ultimo: str = "$"):
    eventos = r.xread({"eventos:stream": ultimo}, count=10, block=5000)
    for _, lista in eventos:
        for eid, campos in lista:
            print(eid, campos)

Geolocalização com GEO

r.geoadd("lojas", (-46.63, -23.55, "sp"), (-43.17, -22.90, "rj"))
proximas = r.geosearch("lojas", longitude=-46.6, latitude=-23.5, radius=50, unit="km")
print(proximas)

HyperLogLog para contagem aproximada

Contar visitantes únicos com pouca memória:

r.pfadd("visitas:hoje", "user1", "user2", "user1")
print(r.pfcount("visitas:hoje"))  # ~2

Transações com pipeline

with r.pipeline() as pipe:
    pipe.set("a", 1)
    pipe.incr("contador")
    pipe.execute()

Configuração de produção do Redis

Em produção, alguns ajustes fazem diferença:

# redis.conf essencial
maxmemory 512mb
maxmemory-policy allkeys-lru
appendonly yes
save 900 1

O appendonly yes ativa persistência AOF (mais durável); o save define snapshots RDB periódicos.

Monitorando o Redis

info = r.info()
print("hit rate:", info["keyspace_hits"] / max(1, info["keyspace_hits"] + info["keyspace_misses"]))
print("memoria usada:", info["used_memory_human"])
print("clientes conectados:", info["connected_clients"])

Um hit rate abaixo de 80% costuma indicar chaves mal dimensionadas ou TTL curto.

Resumo

Redis resolve cache, rate limit, filas e sessões com pouco código. Comece cacheando uma chamada cara, adicione rate limit e depois filas. É uma das ferramentas com melhor retorno de performance por linha de código escrita.