Python e Redis: Cache, Filas e Pub/Sub em 2026

Redis é um dos bancos em memória mais usados no ecossistema Python. Ele aparece como cache, broker para filas, armazenamento de sessões, base para rate limiting, pub/sub e coordenação leve entre serviços. A integração com Python é direta usando redis-py, mas o ganho real vem de saber quando Redis resolve o problema e quando você precisa de uma fila mais completa como Celery, RQ ou Dramatiq.

Este guia foca no uso prático: conectar, cachear, limitar requisições, enfileirar tarefas simples e publicar eventos. Se a sua dúvida principal é tirar trabalho pesado de uma API, leia também FastAPI Background Tasks, Celery e Redis e o verbete de Celery. Redis é uma peça importante dessa arquitetura, mas não deve virar “banco principal disfarçado” nem fila crítica sem estratégia de recuperação.

Quando usar Redis com Python

Use Redis quando você precisa de acesso muito rápido a dados temporários ou estado operacional:

Evite Redis como única fonte de verdade para dados permanentes de produto. Para pedidos, pagamentos, usuários, auditoria e histórico importante, use PostgreSQL, MySQL ou outro banco persistente. Redis entra ao lado deles para reduzir latência, absorver picos e coordenar trabalho temporário.

Instalação e conexão

Instale o cliente Python:

pip install redis

Para rodar Redis localmente com Docker:

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

Conecte usando parâmetros explícitos ou URL:

import redis

# Conexão simples
r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

# Ou via URL, formato comum em deploy
r = redis.from_url("redis://localhost:6379/0", decode_responses=True)

print(r.ping())  # True
print(f"Redis versão: {r.info()['redis_version']}")

O parâmetro decode_responses=True faz o cliente retornar strings em vez de bytes, o que simplifica exemplos, APIs e testes. Em produção, coloque a URL em variável de ambiente e use TLS/senha quando o provedor exigir.

Operações básicas

Redis suporta diferentes estruturas de dados. As mais comuns em aplicações Python são strings, hashes, listas, sets e sorted sets:

import redis

r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

# Strings
r.set("nome", "Ana Silva")
r.set("contador", 0)
print(r.get("nome"))

# Incremento atômico
r.incr("contador")
r.incrby("contador", 10)
print(r.get("contador"))

# Chave com expiração
r.setex("token:sessao:abc", 3600, "usuario_123")
print(r.ttl("token:sessao:abc"))

# Hashes para objetos pequenos
r.hset("usuario:1", mapping={
    "nome": "Ana Silva",
    "email": "[email protected]",
    "plano": "pro",
})
print(r.hgetall("usuario:1"))

# Listas para filas simples
r.rpush("fila:emails", "email_1", "email_2")
print(r.lpop("fila:emails"))

# Sorted sets para ranking ou janelas temporais
r.zadd("ranking", {"ana": 95, "bruno": 87, "carla": 92})
print(r.zrevrange("ranking", 0, -1, withscores=True))

Prefira prefixos claros, como cache:, sessao:, rate_limit: e fila:. Isso facilita debug, limpeza seletiva e métricas.

Cache de funções

O uso mais clássico do Redis é cachear uma operação lenta por alguns minutos:

import hashlib
import json
from functools import wraps

import redis

r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)


def cache_redis(ttl=300):
    """Cache simples para funções puras ou consultas externas."""
    def decorador(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            chave_base = f"{func.__module__}:{func.__name__}"
            args_str = json.dumps({"args": args, "kwargs": kwargs}, sort_keys=True)
            hash_args = hashlib.sha256(args_str.encode()).hexdigest()
            chave = f"cache:{chave_base}:{hash_args}"

            resultado_cache = r.get(chave)
            if resultado_cache is not None:
                return json.loads(resultado_cache)

            resultado = func(*args, **kwargs)
            r.setex(chave, ttl, json.dumps(resultado))
            return resultado
        return wrapper
    return decorador


@cache_redis(ttl=600)
def buscar_dados_api(endpoint):
    """Simula uma consulta lenta em API externa."""
    import time
    time.sleep(2)
    return {"endpoint": endpoint, "status": "ok"}


print(buscar_dados_api("/usuarios"))  # Cache miss
print(buscar_dados_api("/usuarios"))  # Cache hit

Cache só é seguro quando você entende invalidação. Para páginas públicas, relatórios e consultas de catálogo, TTL curto costuma bastar. Para dados sensíveis ou muito mutáveis, invalide a chave no momento da escrita ou prefira não cachear.

Cache em APIs FastAPI

Em APIs, Redis ajuda a reduzir latência de consultas repetidas:

import json

import redis
from fastapi import Depends, FastAPI

app = FastAPI()
cache = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)


async def get_cache():
    return cache


@app.get("/produtos")
async def listar_produtos(r: redis.Redis = Depends(get_cache)):
    chave = "cache:produtos:lista"
    dados_cache = r.get(chave)
    if dados_cache:
        return {"fonte": "cache", "dados": json.loads(dados_cache)}

    produtos = [
        {"id": 1, "nome": "Notebook", "preco": 3500.00},
        {"id": 2, "nome": "Mouse", "preco": 89.90},
        {"id": 3, "nome": "Teclado", "preco": 199.90},
    ]

    r.setex(chave, 300, json.dumps(produtos))
    return {"fonte": "banco", "dados": produtos}


@app.post("/produtos/{produto_id}/atualizar")
async def atualizar_produto(produto_id: int, r: redis.Redis = Depends(get_cache)):
    # Atualize o banco principal antes de invalidar o cache.
    r.delete("cache:produtos:lista")
    return {"status": "atualizado", "produto_id": produto_id}

Esse exemplo é simples. Em uma aplicação real, trate falhas do Redis como degradação: a API pode buscar no banco principal se o cache estiver fora. Cache indisponível não deveria derrubar uma página essencial.

Rate limiting com Redis

Redis é ótimo para controlar taxa de requisições por IP, usuário ou chave de API. Um sorted set permite manter uma janela móvel:

import time


class RateLimiter:
    """Rate limit com janela móvel em Redis."""

    def __init__(self, redis_client, limite, janela_segundos):
        self.r = redis_client
        self.limite = limite
        self.janela = janela_segundos

    def permitir(self, identificador: str) -> bool:
        chave = f"rate_limit:{identificador}"
        agora = time.time()

        pipe = self.r.pipeline()
        pipe.zremrangebyscore(chave, 0, agora - self.janela)
        pipe.zcard(chave)
        pipe.zadd(chave, {str(agora): agora})
        pipe.expire(chave, self.janela)
        _, contagem, _, _ = pipe.execute()

        return contagem < self.limite


limiter = RateLimiter(r, limite=10, janela_segundos=60)

for i in range(12):
    ip = "192.168.1.100"
    status = "permitida" if limiter.permitir(ip) else "bloqueada"
    print(f"Requisição {i + 1}: {status}")

Para produção, considere usar uma biblioteca pronta quando houver requisitos de segurança, múltiplos planos de uso ou integração com gateway/API management. O importante é que a chave tenha escopo correto: limitar por IP pode punir NAT corporativo; limitar por usuário pode ser melhor para APIs autenticadas.

Filas de tarefas: Redis puro, RQ ou Celery?

Redis pode funcionar como fila simples com listas, mas isso não entrega todos os recursos de uma fila robusta. Para um script interno ou protótipo, BLPOP já resolve muita coisa:

import json
from datetime import datetime


def adicionar_tarefa(tipo: str, dados: dict) -> None:
    tarefa = {
        "tipo": tipo,
        "dados": dados,
        "criado_em": datetime.now().isoformat(),
    }
    r.rpush("fila:emails", json.dumps(tarefa))


def consumir_tarefa(timeout=5):
    resultado = r.blpop("fila:emails", timeout=timeout)
    if not resultado:
        return None
    _, tarefa_json = resultado
    return json.loads(tarefa_json)


adicionar_tarefa("enviar_email", {"para": "[email protected]"})
print(consumir_tarefa())

Para trabalho que precisa de retry, histórico, status, múltiplos workers e observabilidade, prefira Celery ou RQ usando Redis como broker. Se a fila processa cobrança, documento fiscal, importação crítica ou sincronização importante, modele idempotência e status no banco principal. O Redis pode guardar a mensagem, mas o produto precisa saber se a tarefa foi recebida, iniciada, concluída, falhou ou deve ser reprocessada.

Pub/Sub para eventos

Pub/Sub permite comunicação em tempo real entre processos, mas não é fila durável. Se um consumidor estiver offline no momento da publicação, ele não recebe a mensagem.

import json


def publicar_evento(canal, evento):
    r.publish(canal, json.dumps(evento))


def ouvir_eventos(canais):
    pubsub = r.pubsub()
    pubsub.subscribe(canais)

    for mensagem in pubsub.listen():
        if mensagem["type"] == "message":
            dados = json.loads(mensagem["data"])
            print(f"[{mensagem['channel']}] {dados}")


publicar_evento("pedidos", {
    "tipo": "novo_pedido",
    "pedido_id": 123,
    "valor": 299.90,
})

Use Pub/Sub para notificações efêmeras, invalidação de cache e sinais internos. Para eventos de negócio que precisam ser auditáveis, prefira uma tabela de outbox, Kafka, RabbitMQ, Redis Streams ou outra solução com persistência e replay.

Sessões com Redis

Sessões são um bom caso de uso porque têm TTL natural:

import json
import uuid
from datetime import datetime


class GerenciadorSessoes:
    """Gerencia sessões temporárias em Redis."""

    def __init__(self, redis_client, ttl=3600):
        self.r = redis_client
        self.ttl = ttl

    def criar(self, usuario_id: str, dados: dict) -> str:
        session_id = str(uuid.uuid4())
        chave = f"sessao:{session_id}"
        sessao = {
            "usuario_id": usuario_id,
            "criada_em": datetime.now().isoformat(),
            **dados,
        }
        self.r.setex(chave, self.ttl, json.dumps(sessao))
        return session_id

    def obter(self, session_id: str) -> dict | None:
        chave = f"sessao:{session_id}"
        dados = self.r.get(chave)
        if dados:
            self.r.expire(chave, self.ttl)
            return json.loads(dados)
        return None

    def destruir(self, session_id: str) -> None:
        self.r.delete(f"sessao:{session_id}")

Não salve senha, token de provedor externo ou dado sensível em texto claro. Mesmo com Redis em rede privada, trate sessão como dado de segurança: defina TTL, use TLS quando necessário e registre apenas o mínimo.

Boas práticas em produção

Ao usar Redis com Python, siga estas recomendações:

• Use decode_responses=True quando quiser strings e JSON simples.
• Defina TTL para caches, sessões, locks e chaves temporárias.
• Estruture nomes de chaves com prefixos previsíveis.
• Use pipelines para operações em lote e reduzir round-trips.
• Configure connection pooling em aplicações com muitas conexões.
• Monitore memória, latência, evictions e taxa de erro do cliente.
• Separe Redis de cache e Redis de fila quando a carga justificar.
• Configure persistência RDB/AOF conforme a criticidade dos dados.
• Evite payloads gigantes em chaves ou mensagens; salve arquivos em storage e envie IDs.
• Combine Redis com logging em Python e OpenTelemetry para entender falhas de cache, fila e worker.

Para carreira, Redis aparece muito em vagas de backend Python, dados e plataforma. Um projeto de portfólio forte pode ser uma API FastAPI que usa Redis para cache, rate limiting e fila de processamento, com Docker Compose, testes com pytest e README explicando decisões. Conecte esse projeto ao guia de portfólio Python e ao checklist de teste técnico Python para transformar tecnologia em evidência profissional.

Conclusão

Redis é uma ferramenta versátil para aplicações Python modernas. Ele acelera consultas, simplifica estado temporário, ajuda a proteger APIs com rate limiting e serve como base para filas e workers. O ponto é escolher o escopo certo: cache pode falhar sem derrubar o produto; fila crítica precisa de retry, idempotência e status; evento de negócio precisa de persistência.

Comece simples, mas nomeie o risco. Se Redis está só evitando uma consulta repetida, TTL e invalidação bastam. Se Redis virou caminho obrigatório de processamento, trate como infraestrutura de produção: monitore, teste falhas, documente recuperação e conecte com uma fila de verdade quando o trabalho não puder se perder.

Redis em alta performance: para aplicações onde o throughput do cliente Redis é crítico, Go com go-redis oferece concorrência nativa com goroutines para processar milhares de operações simultâneas. Rust com a crate redis-rs entrega latência ainda menor para sistemas de cache de missão crítica.