Arquitetura de Software
Arquitetura em Camadas (Layered)
Requisição HTTP
↓
Camada de Apresentação (Routes/Controllers)
↓ valida entrada, chama service
Camada de Negócio (Services)
↓ lógica de negócio, orquestra
Camada de Dados (Repository/DAO)
↓ queries ao banco
Banco de Dados
# Exemplo em FastAPI com camadas
# routes/usuario.py — apresentação
@router.post("/usuarios")
def criar(dados: UsuarioSchema, service: UsuarioService = Depends()):
return service.criar_usuario(dados)
# services/usuario.py — negócio
class UsuarioService:
def __init__(self, repo: UsuarioRepository = Depends()):
self.repo = repo
def criar_usuario(self, dados: UsuarioSchema):
if self.repo.buscar_por_email(dados.email):
raise HTTPException(400, "Email já cadastrado")
senha_hash = bcrypt.hash(dados.senha)
return self.repo.criar(dados.nome, dados.email, senha_hash)
# repositories/usuario.py — dados
class UsuarioRepository:
def __init__(self, db: Session = Depends(get_db)):
self.db = db
def criar(self, nome, email, senha_hash):
usuario = Usuario(nome=nome, email=email, senha=senha_hash)
self.db.add(usuario); self.db.commit()
return usuario
Arquiteturas Comuns
| Arquitetura | Estrutura | Vantagem | Quando usar |
|---|
| Monolito | Tudo em 1 aplicação | Simples de desenvolver | Início de projetos, times pequenos |
| Microsserviços | Serviços independentes | Escala e deploy independentes | Times grandes, alta escala |
| Serverless | Funções sob demanda | Zero gerenciamento de servidor | Eventos, ETL, automações |
| Event-Driven | Comunicação via eventos | Baixo acoplamento | Integração entre sistemas |
⚠ Monolito primeiro!
Comece com monolito bem estruturado. Microsserviços adicionam complexidade enorme (rede, deploy, observabilidade). Só migre quando o monolito se tornar um problema real — não antes.
SOLID na Prática em Python
# S — Single Responsibility
# Ruim: uma classe faz tudo
class UsuarioManager:
def criar(self): ...
def enviar_email(self): ... # responsabilidade extra!
def salvar_log(self): ... # responsabilidade extra!
# Bom: cada classe faz uma coisa
class UsuarioService: def criar(self): ...
class EmailService: def enviar(self): ...
class LogService: def registrar(self): ...
# D — Dependency Inversion (injeta abstração, não implementação)
from abc import ABC, abstractmethod
class NotificacaoBase(ABC):
@abstractmethod
def enviar(self, msg: str): ...
class EmailNotificacao(NotificacaoBase):
def enviar(self, msg): print(f"Email: {msg}")
class SMSNotificacao(NotificacaoBase):
def enviar(self, msg): print(f"SMS: {msg}")
class PedidoService:
def __init__(self, notificacao: NotificacaoBase):
self.notif = notificacao # depende da abstração
def finalizar(self, pedido):
self.notif.enviar(f"Pedido {pedido.id} confirmado!")
Exercício 1: Por que separar Service de Repository?
→ Service cuida da lógica de negócio. Repository cuida do acesso aos dados. São responsabilidades diferentes.
Se trocar PostgreSQL por MongoDB, só muda o Repository — o Service não sabe do banco. Se mudar a regra de negócio, só muda o Service — o banco não sabe. Isso é o princípio de responsabilidade única aplicado em camadas.
Programação Assíncrona e Performance
Async/Await em Python
Python síncrono: enquanto espera o banco responder, a thread fica bloqueada. Python assíncrono: enquanto espera o banco, processa outra requisição. Ideal para APIs com I/O intenso.
import asyncio
import httpx # cliente HTTP async
# Síncrono — bloqueia enquanto espera
def buscar_usuario_sync(id):
resultado = db.query(...) # bloqueia a thread
return resultado
# Assíncrono — libera a thread enquanto espera
async def buscar_usuario(id):
resultado = await db.fetch(...) # libera a thread
return resultado
# FastAPI suporta async nativamente
@app.get("/usuarios/{id}")
async def get_usuario(id: int):
usuario = await usuario_service.buscar(id)
return usuario
# Executar múltiplas operações em paralelo
async def dados_dashboard(user_id: int):
# executa as 3 queries ao mesmo tempo!
pedidos, saldo, notifs = await asyncio.gather(
buscar_pedidos(user_id),
buscar_saldo(user_id),
buscar_notificacoes(user_id)
)
return {"pedidos": pedidos, "saldo": saldo, "notifs": notifs}
⚠ Quando usar async
Use async para
operações de I/O: banco de dados, APIs externas, leitura de arquivos. Para operações
CPU-intensivas (processamento de imagem, ML), async não ajuda — use multiprocessing ou workers separados.
Otimização de Queries — N+1 Problem
# PROBLEMA N+1 — faz 1 query + N queries (uma por usuário)
usuarios = db.query(Usuario).all()
for u in usuarios:
print(u.pedidos) # gera 1 query por usuario!
# SOLUÇÃO — eager loading (1 query com JOIN)
usuarios = db.query(Usuario).options(
joinedload(Usuario.pedidos)
).all()
# Agora: 1 query só
# Paginação — nunca retorne tudo!
@app.get("/usuarios")
async def listar(pagina: int = 1, limite: int = 20):
offset = (pagina - 1) * limite
return db.query(Usuario).offset(offset).limit(limite).all()
# Índices — essencial para queries rápidas
# No SQLAlchemy:
class Usuario(Base):
email = Column(String, index=True) # cria índice automaticamente
Exercício 2: O que é o problema N+1 e como resolver?
→ N+1 ocorre quando se faz 1 query para listar + N queries para buscar detalhes de cada item. Resolver com eager loading (JOIN na query principal).
100 usuários com pedidos = 1 + 100 = 101 queries. Com joinedload = 1 query com JOIN. A diferença pode ser de milissegundos vs segundos em produção.
Cache e Filas de Mensagens
Cache com Redis
Cache armazena resultados computados para evitar reprocessamento. Redis é o cache mais usado — in-memory, extremamente rápido (sub-milissegundo).
pip install redis
import redis
import json
r = redis.Redis(host='localhost', port=6379, db=0)
# Cache simples
def buscar_produto(produto_id: int):
cache_key = f"produto:{produto_id}"
# Tentar do cache primeiro
cached = r.get(cache_key)
if cached:
return json.loads(cached) # cache hit!
# Cache miss — busca do banco
produto = db.query(Produto).get(produto_id)
if produto:
# Salva no cache por 5 minutos
r.setex(cache_key, 300, json.dumps(produto.dict()))
return produto
# Decorator de cache (mais elegante)
from functools import wraps
def cache(expira_em=300):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
key = f"{func.__name__}:{args}:{kwargs}"
cached = r.get(key)
if cached: return json.loads(cached)
resultado = await func(*args, **kwargs)
r.setex(key, expira_em, json.dumps(resultado))
return resultado
return wrapper
return decorator
@cache(expira_em=600)
async def ranking_produtos():
return await db.fetch("SELECT * FROM produtos ORDER BY vendas DESC LIMIT 10")
Filas com Celery + Redis
Filas de mensagens processam tarefas pesadas em background sem bloquear a resposta da API. Ex: enviar email, gerar PDF, processar imagem — o usuário não espera.
pip install celery redis
# celery_app.py
from celery import Celery
app = Celery('tasks', broker='redis://localhost:6379/0')
@app.task
def enviar_email(destinatario: str, assunto: str, corpo: str):
# lógica de envio de email
smtp.send(destinatario, assunto, corpo)
print(f"Email enviado para {destinatario}")
@app.task
def gerar_relatorio(usuario_id: int):
dados = buscar_dados(usuario_id)
pdf = criar_pdf(dados)
salvar_s3(pdf)
# Na API — não bloqueia a resposta!
@app.post("/usuarios/{id}/relatorio")
async def solicitar_relatorio(id: int):
gerar_relatorio.delay(id) # enfileira a tarefa
return {"mensagem": "Relatório sendo gerado, você receberá por email"}
# Rodar worker em outro terminal:
# celery -A celery_app worker --loglevel=info
| Tecnologia | Uso | Quando usar |
|---|
| Redis (cache) | Armazenar resultados temporários | Queries lentas, dados que não mudam muito |
| Celery + Redis | Processar tarefas em background | Email, PDF, processamento de imagem |
| RabbitMQ | Fila de mensagens robusta | Integração entre sistemas, alta confiabilidade |
| Kafka | Stream de eventos em alta escala | Logs, analytics em tempo real, big data |
Exercício 3: Por que usar fila para envio de email e não fazer direto na API?
→ Envio de email é lento (rede, SMTP) e pode falhar. Com fila, a API responde imediatamente e o email é enviado em background com retry automático.
Sem fila: usuário espera 2-3 segundos pela resposta da API enquanto o email é enviado. Se falhar, o usuário recebe erro. Com Celery: API responde em ms, Celery tenta enviar o email e, se falhar, tenta novamente automaticamente.
Microsserviços e Comunicação
Quando migrar para Microsserviços
Sinais que o monolito está no limite
Deploy demora muito e afeta tudo.
Times diferentes editam os mesmos arquivos.
Um módulo lento afeta toda a aplicação.
Precisam de tecnologias diferentes por módulo.
Benefícios dos microsserviços
Deploy independente por serviço.
Escala só o que precisa (ex: só o serviço de busca).
Times autônomos com tecnologias diferentes.
Falha isolada — um serviço cai sem derrubar tudo.
Comunicação entre Microsserviços
| Tipo | Como | Quando usar | Problema |
|---|
| Síncrona (REST/gRPC) | Serviço A chama API do B e espera | Resultado imediato necessário | Acoplamento, se B cair, A falha |
| Assíncrona (Mensagens) | Publica evento na fila, B consome depois | Operações que podem ser atrasadas | Consistência eventual |
# API Gateway — ponto de entrada único
# Nginx como gateway simples
upstream servico_usuarios {
server usuarios-api:8001;
}
upstream servico_pedidos {
server pedidos-api:8002;
}
server {
location /api/usuarios { proxy_pass http://servico_usuarios; }
location /api/pedidos { proxy_pass http://servico_pedidos; }
}
# gRPC — comunicação eficiente entre serviços internos
# Proto file: usuario.proto
syntax = "proto3";
service UsuarioService {
rpc BuscarUsuario (UsuarioRequest) returns (UsuarioResponse);
}
message UsuarioRequest { int32 id = 1; }
message UsuarioResponse { string nome = 1; string email = 2; }
Observabilidade — Saber o que está acontecendo
import logging
import time
from opentelemetry import trace
# Logging estruturado
logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)
@app.middleware("http")
async def log_requests(request, call_next):
inicio = time.time()
response = await call_next(request)
duracao = time.time() - inicio
logger.info(f"method={request.method} path={request.url.path} "
f"status={response.status_code} duration={duracao:.3f}s")
return response
# Métricas com Prometheus
from prometheus_client import Counter, Histogram
requisicoes = Counter('http_requests_total', 'Total de requisições', ['method', 'path'])
latencia = Histogram('http_request_duration_seconds', 'Latência das requisições')
⚠ Os 3 pilares de Observabilidade
Logs: o que aconteceu (eventos).
Métricas: quantas vezes, quanto tempo (números).
Traces: o caminho de uma requisição entre serviços. Ferramentas: ELK Stack, Prometheus+Grafana, Jaeger.
📝 Cola — Técnicas Avançadas
Arquitetura em Camadas
Routes → Services → Repositories → Database
(entrada) (negócio) (dados) (persistência)
Async vs Sync
| Situação | Use |
|---|
| I/O: banco, API externa, arquivo | async/await |
| CPU: ML, processamento de imagem | multiprocessing / worker |
| Múltiplas operações independentes | asyncio.gather() |
⭐ Regras de Ouro
Regra
Comece com
monolito bem estruturado — microsserviços são complexidade, não bala de prata
Regra
N+1: sempre use
eager loading em relações. Nunca query dentro de loop
Regra
Cache: leia do Redis primeiro. Se miss, busca do banco e
salva no cacheRegraTarefas lentas (email, PDF) → fila Celery. API responde imediatamente
RegraService não conhece o banco. Repository não conhece regra de negócio
Regra
Observabilidade:
logs + métricas + traces — sem isso você voa cego em produção