-- Window Functions — cálculos sobre partições
SELECT
nome,
departamento,
salario,
AVG(salario) OVER (PARTITION BY departamento) AS media_depto,
RANK() OVER (PARTITION BY departamento ORDER BY salario DESC) AS ranking,
ROW_NUMBER() OVER (ORDER BY salario DESC) AS posicao_global
FROM funcionarios;
-- CTE (Common Table Expression) — queries legíveis
WITH vendas_mensais AS (
SELECT
DATE_TRUNC('month', criado_em) AS mes,
SUM(total) AS receita
FROM pedidos
WHERE status = 'pago'
GROUP BY 1
),
crescimento AS (
SELECT
mes,
receita,
LAG(receita) OVER (ORDER BY mes) AS receita_anterior,
ROUND((receita / LAG(receita) OVER (ORDER BY mes) - 1) * 100, 2) AS crescimento_pct
FROM vendas_mensais
)
SELECT * FROM crescimento ORDER BY mes;
-- Subquery vs JOIN — saiba quando usar cada um
-- Subquery correlacionada (lenta — executada N vezes):
SELECT nome FROM usuarios u
WHERE (SELECT COUNT(*) FROM pedidos WHERE usuario_id = u.id) > 5;
-- JOIN equivalente (mais rápido):
SELECT u.nome FROM usuarios u
JOIN (
SELECT usuario_id, COUNT(*) as qtd
FROM pedidos GROUP BY 1
) p ON p.usuario_id = u.id AND p.qtd > 5;
Recursos Avançados do PostgreSQL
-- JSONB — armazenar JSON com indexação
CREATE TABLE eventos (
id SERIAL PRIMARY KEY,
dados JSONB NOT NULL,
criado_em TIMESTAMPTZ DEFAULT NOW()
);
INSERT INTO eventos (dados) VALUES
('{"tipo": "login", "usuario_id": 42, "ip": "192.168.1.1"}');
-- Buscar por campo JSON
SELECT * FROM eventos WHERE dados->>'tipo' = 'login';
SELECT * FROM eventos WHERE (dados->>'usuario_id')::int = 42;
-- Índice em campo JSONB
CREATE INDEX idx_eventos_tipo ON eventos USING GIN (dados);
-- Arrays nativos
CREATE TABLE tags_produto (
produto_id INT,
tags TEXT[]
);
SELECT * FROM tags_produto WHERE 'eletrônico' = ANY(tags);
-- Full Text Search
SELECT * FROM artigos
WHERE to_tsvector('portuguese', conteudo) @@ to_tsquery('python & backend');
Exercício 1: Quando usar CTE em vez de subquery?
→ CTE quando a query é reutilizada ou precisa de mais legibilidade. Subquery para casos simples de uma vez só.
CTEs tornam queries complexas legíveis — cada WITH é um "bloco nomeado". O PostgreSQL pode materializá-las (calcular uma vez e reutilizar). Subqueries correlacionadas são geralmente lentas pois executam N vezes.
SQLAlchemy Avançado e Alembic
SQLAlchemy — Relacionamentos e Queries
from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, func
from sqlalchemy.orm import relationship, DeclarativeBase
class Base(DeclarativeBase): pass
class Usuario(Base):
__tablename__ = "usuarios"
id = Column(Integer, primary_key=True)
nome = Column(String(100), nullable=False)
email = Column(String(100), unique=True, index=True)
criado = Column(DateTime, default=func.now())
pedidos = relationship("Pedido", back_populates="usuario",
cascade="all, delete-orphan")
class Pedido(Base):
__tablename__ = "pedidos"
id = Column(Integer, primary_key=True)
total = Column(Integer, nullable=False) # em centavos!
usuario_id = Column(Integer, ForeignKey("usuarios.id"), nullable=False)
usuario = relationship("Usuario", back_populates="pedidos")
itens = relationship("ItemPedido", back_populates="pedido")
# Queries avançadas com SQLAlchemy
from sqlalchemy import select, func, and_, or_
# Agregação com group by
stmt = (
select(Usuario.nome, func.count(Pedido.id).label("total_pedidos"),
func.sum(Pedido.total).label("valor_total"))
.join(Pedido, isouter=True) # LEFT JOIN
.group_by(Usuario.id)
.having(func.count(Pedido.id) > 0)
.order_by(func.sum(Pedido.total).desc())
)
resultado = session.execute(stmt).all()
# Filtros dinâmicos
def buscar_usuarios(nome=None, email=None, ativo=None):
filtros = []
if nome: filtros.append(Usuario.nome.ilike(f"%{nome}%"))
if email: filtros.append(Usuario.email == email)
if ativo is not None: filtros.append(Usuario.ativo == ativo)
return session.query(Usuario).filter(and_(*filtros)).all()
Alembic — Migrations de Banco
pip install alembic
# Inicializar
alembic init alembic
# alembic/env.py — configurar o modelo
from models import Base
target_metadata = Base.metadata
# Criar migration automaticamente (detecta mudanças no modelo)
alembic revision --autogenerate -m "add campo telefone em usuarios"
# Arquivo gerado: alembic/versions/abc123_add_campo_telefone.py
def upgrade():
op.add_column('usuarios', sa.Column('telefone', sa.String(20)))
def downgrade():
op.drop_column('usuarios', 'telefone')
# Aplicar migrations
alembic upgrade head # aplica todas pendentes
alembic upgrade +1 # aplica próxima
alembic downgrade -1 # desfaz última
alembic current # versão atual
alembic history # histórico
alembic downgrade base # desfaz tudo (cuidado!)
⚠ Nunca altere migration já aplicada em produção
Se precisar mudar o banco, crie uma nova migration. Alterar uma migration já aplicada quebra o histórico e outros desenvolvedores não saberão o que está no banco deles. Migrations são como commits — imutáveis após push.
Exercício 2: Por que usar migrations em vez de alterar o banco manualmente?
→ Migrations são versionadas, rastreáveis e reproduzíveis — todos os ambientes ficam iguais.
Sem migration: "funcionou na minha máquina mas não em produção" porque alguém esqueceu de rodar um ALTER TABLE. Com Alembic: alembic upgrade head garante que todos os ambientes estão na mesma versão do schema.
Otimização de Queries e Índices
EXPLAIN ANALYZE — Entender a query
-- Ver o plano de execução de uma query
EXPLAIN ANALYZE
SELECT u.nome, COUNT(p.id) as total
FROM usuarios u
LEFT JOIN pedidos p ON p.usuario_id = u.id
WHERE u.criado_em > '2024-01-01'
GROUP BY u.id;
-- Resultado típico:
-- Hash Aggregate (cost=250.00..260.00 rows=1000)
-- -> Hash Left Join (cost=50.00..200.00)
-- Hash Cond: (p.usuario_id = u.id)
-- -> Seq Scan on pedidos ← SEM ÍNDICE! lento
-- -> Index Scan on usuarios ← COM ÍNDICE! rápido
-- Seq Scan = percorre a tabela toda (ruim para tabelas grandes)
-- Index Scan = usa índice (muito mais rápido)
Tipos de Índice no PostgreSQL
Tipo
Uso
Exemplo
B-tree (padrão)
Igualdade e range (<, >, BETWEEN)
WHERE id = 42 ou WHERE data > '2024'
Hash
Só igualdade — mais rápido que B-tree
WHERE email = 'ana@ex.com'
GIN
Arrays, JSONB, full-text search
WHERE tags @> ARRAY['python']
GiST
Dados geométricos, ranges
Distância entre coordenadas
Parcial
Só parte das linhas
WHERE ativo = true (indexa só ativos)
-- Criar índices estratégicos
CREATE INDEX idx_pedidos_usuario ON pedidos (usuario_id);
CREATE INDEX idx_pedidos_status ON pedidos (status) WHERE status != 'cancelado'; -- parcial
CREATE INDEX idx_usuarios_email ON usuarios (email); -- já tem unique, mas explicita
CREATE INDEX idx_pedidos_data_total ON pedidos (criado_em, total); -- composto
-- Índice composto: útil quando WHERE usa ambas as colunas juntas
-- WHERE criado_em > '2024' AND total > 10000
-- Ver índices existentes
SELECT indexname, indexdef FROM pg_indexes WHERE tablename = 'pedidos';
-- Ver índices não usados (candidatos a remoção)
SELECT schemaname, tablename, indexname
FROM pg_stat_user_indexes
WHERE idx_scan = 0 -- nunca usado!
Connection Pooling
pip install asyncpg sqlalchemy[asyncio]
# Sem pool: cada request abre e fecha conexão com o banco (lento!)
# Com pool: conexões são reutilizadas
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
engine = create_async_engine(
"postgresql+asyncpg://user:pass@localhost/db",
pool_size=20, # conexões permanentes no pool
max_overflow=10, # conexões extras sob alta carga
pool_pre_ping=True, # testa conexão antes de usar
pool_recycle=3600, # recicla conexões a cada 1h
)
# Em produção: use PgBouncer para pooling externo
# PgBouncer suporta milhares de conexões de app → poucos do banco
Exercício 3: Quando um índice pode ser prejudicial?
→ Em tabelas com muitas escritas (INSERT/UPDATE/DELETE) — o índice precisa ser atualizado a cada operação.
Índice em coluna de uma tabela de logs (milhões de inserts por dia) vai degradar a escrita. Crie índices só em colunas frequentemente lidas em WHERE, JOIN e ORDER BY. Use EXPLAIN ANALYZE para confirmar que o índice está sendo usado.
Redis Avançado e Padrões de Banco
Estruturas de Dados do Redis
import redis
r = redis.Redis(host='localhost', port=6379)
# String — cache simples, contadores
r.set("usuario:42:nome", "Ana Silva")
r.get("usuario:42:nome") # b'Ana Silva'
r.incr("visitas:home") # contador atômico
r.expire("session:abc", 1800) # TTL de 30min
# Hash — objeto com múltiplos campos
r.hset("usuario:42", mapping={
"nome": "Ana", "email": "ana@ex.com", "pontos": 150
})
r.hget("usuario:42", "nome") # b'Ana'
r.hgetall("usuario:42") # todos os campos
r.hincrby("usuario:42", "pontos", 10) # +10 pontos
# List — fila ou pilha
r.rpush("fila:emails", "ana@ex.com") # enqueue
r.rpush("fila:emails", "bia@ex.com")
r.llen("fila:emails") # 2
r.lpop("fila:emails") # dequeue: b'ana@ex.com'
# Set — conjunto único
r.sadd("online:usuarios", 42, 77, 99)
r.sismember("online:usuarios", 42) # True
r.smembers("online:usuarios") # {42, 77, 99}
r.scard("online:usuarios") # 3
# Sorted Set — ranking com pontuação
r.zadd("ranking", {"Ana": 1500, "Bia": 2300, "Carlos": 800})
r.zrange("ranking", 0, -1, withscores=True, rev=True)
# [(b'Bia', 2300), (b'Ana', 1500), (b'Carlos', 800)]
Padrões de Banco Avançados
CQRS — Command Query Responsibility Segregation
Separa escrita (Command) de leitura (Query). Banco de escrita normalizado. Banco de leitura desnormalizado/cache otimizado para leitura rápida.
Event Sourcing
Armazena eventos em vez do estado atual. Estado = replay de todos os eventos. Ex: em vez de saldo atual, guarda cada crédito/débito.
Soft Delete
Em vez de DELETE, marca registro como inativo. deleted_at = Column(DateTime, nullable=True) Preserva histórico e facilita restauração.
Audit Trail
Registra todas as alterações em tabelas críticas. Quem alterou, quando e o que mudou. Essencial para conformidade (LGPD, auditorias).
-- Soft Delete + Audit Trail no PostgreSQL
-- Tabela com soft delete
CREATE TABLE usuarios (
id SERIAL PRIMARY KEY,
nome TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
criado_em TIMESTAMPTZ DEFAULT NOW(),
atualizado TIMESTAMPTZ DEFAULT NOW(),
deletado TIMESTAMPTZ -- NULL = ativo, preenchido = deletado
);
-- View que filtra deletados automaticamente
CREATE VIEW usuarios_ativos AS
SELECT * FROM usuarios WHERE deletado IS NULL;
-- Trigger para audit trail
CREATE TABLE audit_log (
id SERIAL PRIMARY KEY,
tabela TEXT,
operacao TEXT, -- INSERT / UPDATE / DELETE
usuario TEXT,
timestamp TIMESTAMPTZ DEFAULT NOW(),
dados_antes JSONB,
dados_depois JSONB
);
CREATE OR REPLACE FUNCTION audit_trigger() RETURNS TRIGGER AS $$
BEGIN
INSERT INTO audit_log (tabela, operacao, usuario, dados_antes, dados_depois)
VALUES (TG_TABLE_NAME, TG_OP, current_user,
row_to_json(OLD), row_to_json(NEW));
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER audit_usuarios
AFTER INSERT OR UPDATE OR DELETE ON usuarios
FOR EACH ROW EXECUTE FUNCTION audit_trigger();
📝 Cola — Lab de Banco de Dados
SQL Avançado — Referência Rápida
-- Window Function
RANK() OVER (PARTITION BY coluna ORDER BY outra)
-- CTE
WITH nome AS (SELECT ...) SELECT * FROM nome;
-- EXPLAIN ANALYZE — ver plano de execução
EXPLAIN ANALYZE SELECT ...;
-- Seq Scan = ruim para tabelas grandes → precisa de índice
-- Index Scan = usando índice → rápido
Estruturas Redis × Caso de Uso
Estrutura
Use para
String
Cache simples, contadores, sessões
Hash
Objetos com campos (perfil de usuário)
List
Fila (rpush + lpop) ou pilha (rpush + rpop)
Set
Usuários online, tags únicas, interseções
Sorted Set
Rankings, leaderboards, filas com prioridade
⭐ Regras de Ouro
Regra
EXPLAIN ANALYZE antes de criar índice — confirme que ele será usado
Regra
Nunca altere migration já aplicada — crie uma nova migration
Regra
Soft delete preserva histórico — deleted_at em vez de DELETE real
RegraValores monetários em centavos (int) — nunca float (problema de precisão)
Regra
Pool de conexões é obrigatório em produção — pool_size mínimo 10
Regra
Índice não usado = overhead de escrita sem benefício — remova índices sem uso