 |
| Diagrama de arquitetura microserviços FastAPI: Auth, Product, Sales com SQLite e chamadas HTTP entre serviços |
Introdução
Neste tutorial você vai aprender a construir uma arquitetura de micro-APIs similar ao diagrama:-
Auth API: gera e valida tokens JWT;
-
Product API: CRUD de produtos com SQLite;
-
Sales API: registra vendas (SQLite) e faz chamadas para outras APIs (ex: validação de produto);
Tudo sem containers, ideal para desenvolvimento local e testes rápidos. Tudo em FastAPI, com exemplos prontos para rodar em portas diferentes.
Por que essa arquitetura?
-
Permite separar responsabilidades (autenticação, catálogo, vendas).
-
Cada serviço pode escalar de forma independente no futuro.
-
Fácil evolução para filas assíncronas (RabbitMQ) ou bancos separados (Postgres/Mongo) quando desejar.
Requisitos (instalação rápida)
No seu ambiente Python (recomendo criar um venv):
python -m venv venv
source venv/bin/activate # macOS / Linux
venv\Scripts\activate # Windows
pip install fastapi uvicorn sqlalchemy pydantic jwt passlib[bcrypt] requests
Arquivo requirements.txt sugerido:
fastapi
uvicorn[standard]
sqlalchemy
pydantic
PyJWT
passlib[bcrypt]
requests
microservices-fastapi/├── auth_api/│ └── main.py├── product_api/│ └── main.py├── sales_api/│ └── main.py└── README.md
1) Auth API — gerar e validar JWT
Objetivo: emitir token JWT ao autenticar usuário (aqui exemplo simples com usuário hardcoded para demo).
Arquivo: auth_api/main.py
# auth_api/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import jwt, datetime
from passlib.context import CryptContext
app = FastAPI(title="Auth API")
SECRET_KEY = "troque_isto_por_um_segredo_forte"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# Exemplo simples: usuário fixo (em produção, use DB)
fake_user = {
"username": "admin",
"hashed_password": pwd_context.hash("123")
}
class LoginIn(BaseModel):
username: str
password: str
@app.post("/login")
def login(payload: LoginIn):
if payload.username != fake_user["username"] or not pwd_context.verify(payload.password, fake_user["hashed_password"]):
raise HTTPException(status_code=401, detail="Usuário ou senha incorretos")
expire = datetime.datetime.utcnow() + datetime.timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
token = jwt.encode({"sub": payload.username, "exp": expire}, SECRET_KEY, algorithm=ALGORITHM)
return {"access_token": token, "token_type": "bearer"}
uvicorn auth_api.main:app --host 0.0.0.0 --port 8000 --reload
curl -X POST "http://localhost:8000/login" -H "Content-Type: application/json" -d '{"username":"admin","password":"123"}'
2) Product API — CRUD com SQLite
Objetivo: fornecer endpoints para criar/listar produtos; protegido por JWT.
Arquivo: product_api/main.py
# product_api/main.py
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.orm import sessionmaker, declarative_base
import jwt
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import requests
DATABASE_URL = "sqlite:///./product.db"
SECRET_KEY = "troque_isto_por_um_segredo_forte" # o mesmo do Auth em demo
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()
class Product(Base):
__tablename__ = "products"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
price = Column(Float)
Base.metadata.create_all(bind=engine)
app = FastAPI(title="Product API")
security = HTTPBearer()
class ProductIn(BaseModel):
name: str
price: float
def verify_token(creds: HTTPAuthorizationCredentials = Depends(security)):
token = creds.credentials
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
except jwt.PyJWTError:
raise HTTPException(status_code=401, detail="Token inválido")
return payload
@app.post("/products")
def create_product(prod: ProductIn, payload=Depends(verify_token)):
db = SessionLocal()
p = Product(name=prod.name, price=prod.price)
db.add(p)
db.commit()
db.refresh(p)
db.close()
return {"id": p.id, "name": p.name, "price": p.price}
@app.get("/products")
def list_products(skip: int = 0, limit: int = 100, payload=Depends(verify_token)):
db = SessionLocal()
items = db.query(Product).offset(skip).limit(limit).all()
db.close()
return items
uvicorn product_api.main:app --host 0.0.0.0 --port 8001 --reload
Fluxo de teste (usar token do Auth):
-
Obter token no Auth (
/login). -
Criar produto:
curl -X POST "http://localhost:8001/products" \-H "Authorization: Bearer <TOKEN>" \-H "Content-Type: application/json" \-d '{"name":"Mouse","price":29.9}'
curl -H "Authorization: Bearer <TOKEN>" "http://localhost:8001/products"
3) Sales API — registrar venda e validar produto via Product API
Objetivo: registrar vendas localmente em SQLite e validar produto consultando Product API (HTTP).
Arquivo: sales_api/main.py
# sales_api/main.py
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime
from sqlalchemy.orm import sessionmaker, declarative_base
import datetime
import requests
DATABASE_URL = "sqlite:///./sales.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()
class Sale(Base):
__tablename__ = "sales"
id = Column(Integer, primary_key=True, index=True)
product_id = Column(Integer)
product_name = Column(String)
quantity = Column(Integer)
total = Column(Float)
created_at = Column(DateTime, default=datetime.datetime.utcnow)
Base.metadata.create_all(bind=engine)
app = FastAPI(title="Sales API")
class SaleIn(BaseModel):
product_id: int
quantity: int
PRODUCT_API_URL = "http://localhost:8001/products" # endpoint base
@app.post("/sales")
def create_sale(sale: SaleIn):
# Consulta Product API para validar produto (simples: buscar lista e filtrar)
resp = requests.get(PRODUCT_API_URL)
if resp.status_code != 200:
raise HTTPException(status_code=502, detail="Product API indisponível")
products = resp.json()
product = next((p for p in products if p["id"] == sale.product_id), None)
if not product:
raise HTTPException(status_code=404, detail="Produto não encontrado")
total = product["price"] * sale.quantity
db = SessionLocal()
s = Sale(product_id=product["id"], product_name=product["name"], quantity=sale.quantity, total=total)
db.add(s)
db.commit()
db.refresh(s)
db.close()
# Aqui você poderia publicar uma mensagem na fila RabbitMQ (opcional)
return {"sale_id": s.id, "product": s.product_name, "total": s.total}
uvicorn sales_api.main:app --host 0.0.0.0 --port 8002 --reload
Testando (exemplo):
-
Criar produto via Product API (veja acima).
-
Criar venda:
curl -X POST "http://localhost:8002/sales" -H "Content-Type: application/json" -d '{"product_id":1,"quantity":2}'
Comunicação entre serviços (HTTP calls)
No exemplo da Sales API usamos requests.get para consultar GET /products e validar o produto. Em arquiteturas reais, você pode:
-
Usar chamadas HTTP síncronas (como no exemplo) — simples, mas acoplamento temporal.
-
Mudar para mensageria (RabbitMQ) para eventos assíncronos (venda registrada → event published).
-
Implementar retries, circuit breaker e timeouts (ex.:
requests.get(url, timeout=3)).
Boas práticas e melhorias recomendadas
-
Variáveis de ambiente: mova segredos (SECRET_KEY) e URLs para variáveis (use
python-decoupleoupydantic.Settings). -
Migrations: use Alembic para gerenciar schema do SQLAlchemy em produção.
-
Autenticação real: troque usuário hardcoded por tabela
usersno DB e endpoint de registro. -
HTTPS em produção (Nginx/Traefik + certificado).
-
Logs estruturados e monitoramento (Prometheus/Grafana).
-
Testes automatizados: unit + integration tests com
pytestehttpx.
Erros comuns e tratamento
-
Token expirado: implemente refresh token (rota
/refresh). -
Product API indisponível: adicione fallback ou fila para processar vendas offline.
-
SQLite limita concorrência: para produção use PostgreSQL (mais robusto para multi-conexões).
Conclusão (call to action)
Com esse guia você tem uma base funcional para montar as três APIs em FastAPI com SQLite, imitando a arquitetura do seu diagrama. É um ponto de partida rápido para testes locais e evolução para um ambiente em containers ou nuvem.
👉 Próximos passos sugeridos:
-
Migrar Product e Sales para PostgreSQL/Mongo;
-
Adicionar RabbitMQ para comunicação assíncrona;
-
Implementar testes automáticos e CI/CD.
Recursos adicionais e links úteis
-
FastAPI — https://fastapi.tiangolo.com
-
SQLAlchemy — https://www.sqlalchemy.org
-
PyJWT — https://pyjwt.readthedocs.io
-
uvicorn — https://www.uvicorn.org
