Início » API » 🔐 Construindo 3 Micro-APIs em Python com FastAPI (Auth, Product, Sales) — Arquitetura, JWT e SQLite

📑 Tabela de Conteúdo

🔐 Construindo 3 Micro-APIs em Python com FastAPI (Auth, Product, Sales) — Arquitetura, JWT e SQLite


 

Diagrama de arquitetura microserviços FastAPI: Auth, Product, Sales com SQLite e chamadas HTTP entre serviços
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

Estrutura de pastas (sugestão)

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"}

Executar:
uvicorn auth_api.main:app --host 0.0.0.0 --port 8000 --reload
Teste:
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

Executar:
uvicorn product_api.main:app --host 0.0.0.0 --port 8001 --reload

Fluxo de teste (usar token do Auth):

  1. Obter token no Auth (/login).

  2. Criar produto:

curl -X POST "http://localhost:8001/products" \
 -H "Authorization: Bearer <TOKEN>" \
 -H "Content-Type: application/json" \
 -d '{"name":"Mouse","price":29.9}'
Listar:
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}

Executar:
uvicorn sales_api.main:app --host 0.0.0.0 --port 8002 --reload

Testando (exemplo):

  1. Criar produto via Product API (veja acima).

  2. 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-decouple ou pydantic.Settings).

  • Migrations: use Alembic para gerenciar schema do SQLAlchemy em produção.

  • Autenticação real: troque usuário hardcoded por tabela users no DB e endpoint de registro.

  • HTTPS em produção (Nginx/Traefik + certificado).

  • Logs estruturados e monitoramento (Prometheus/Grafana).

  • Testes automatizados: unit + integration tests com pytest e httpx.

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

Sobre o Autor

iHoney

Klebes Santos
Sou um jovem apaixonado por tecnologia e sempre em busca de aprendizado. Gosto de explorar novas ferramentas, linguagens de programação e tendências do mundo digital. Trabalho com desenvolvimento de software, especialmente em Java e Python, e tenho experiência com bancos de dados e integração de sistemas. Além disso, me interesso por automação, APIs e otimização de processos. No dia a dia, gosto de solucionar problemas e encontrar maneiras eficientes de melhorar aplicações. Também compartilho conhecimento através do meu blog, onde escrevo sobre tecnologia e programação. Estou sempre disposto a aprender e evoluir, acompanhando as inovações do setor.

✝ Copyright © Blog do KDS - Isaías 40:5 “A glória do Senhor se manifestará, e toda a humanidade a verá.”