--- title: "Python e Redis: Cache, Filas e Pub/Sub em 2026" url: "https://python.dev.br/blog/python-e-redis/" markdown_url: "https://python.dev.br/blog/python-e-redis.MD" description: "Aprenda a usar Redis com Python para cache, filas simples, rate limiting, sessões e pub/sub, com exemplos práticos, cuidados de produção e links para Celery e FastAPI." date: "2026-01-28" author: "Equipe Python Brasil" --- # Python e Redis: Cache, Filas e Pub/Sub em 2026 Aprenda a usar Redis com Python para cache, filas simples, rate limiting, sessões e pub/sub, com exemplos práticos, cuidados de produção e links para Celery e FastAPI. 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](/blog/fastapi-background-tasks-celery-redis-2026/) e o verbete de [Celery](/glossario/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: | Cenário | Redis ajuda como | |---|---| | Cache de resposta de API | Chave com TTL e invalidação controlada | | Rate limiting | Contadores ou sorted sets por IP, usuário ou token | | Sessão curta | Dados temporários com expiração automática | | Fila simples | Lista, stream, RQ ou broker de Celery | | Evento em tempo real | Pub/Sub entre processos | | Lock leve | Chave temporária com expiração | 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: ```bash pip install redis ``` Para rodar Redis localmente com Docker: ```bash docker run -d --name redis -p 6379:6379 redis:7-alpine ``` Conecte usando parâmetros explícitos ou URL: ```python 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: ```python 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": "ana@example.com", "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: ```python 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: ```python 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: ```python 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: ```python 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": "ana@example.com"}) print(consumir_tarefa()) ``` Para trabalho que precisa de retry, histórico, status, múltiplos workers e observabilidade, prefira [Celery](/glossario/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. ```python 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: ```python 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](/blog/logging-em-python/) e [OpenTelemetry](/blog/opentelemetry-python-observabilidade/) 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](/guias/testes-com-pytest/) e README explicando decisões. Conecte esse projeto ao guia de [portfólio Python](/carreira/projetos-portfolio-python/) e ao checklist de [teste técnico Python](/carreira/teste-tecnico-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.