📚 Módulo 03: Backend - Schemas Pydantic (DTOs) e Exceções Globais
Em Engenharia de Software, as APIs corporativas devem blindar o banco de dados contra inserção de dados maliciosos ou corruptos e ocultar do usuário detalhes internos da infraestrutura. Neste módulo, aprenderemos a implementar a barreira de isolamento de dados com Schemas Pydantic v2 (Padrão DTO) e a programar um Handler Global de Exceções para padronizar todos os retornos de erros.
🛡️ 1. O Padrão DTO com Pydantic v2
No ecossistema FastAPI, o Pydantic é a biblioteca padrão para validação, tipagem e serialização de dados. Os schemas do Pydantic atuam como DTOs (Data Transfer Objects), equivalentes diretos aos Java Records.
💻 1. Schemas de Produto (Entrada e Saída)
Crie o arquivo em app/schemas/produto_schema.py:
from decimal import Decimal
from typing import Optional
from pydantic import BaseModel, Field, field_validator
# DTO de Entrada: Valida dados enviados pelo usuário
class ProdutoFormSchema(BaseModel):
nome: str = Field(..., min_length=3, max_length=150, description="Nome do eletrônico")
descricao: Optional[str] = Field(None, max_length=500, description="Ficha técnica")
preco: Decimal = Field(..., gt=Decimal("0.00"), description="Preço deve ser maior que zero")
estoque: int = Field(..., ge=0, description="Estoque não pode ser negativo")
categoria_id: int = Field(..., ge=1, description="ID de categoria válido")
@field_validator("preco")
def validar_preco_positivo(cls, v):
if v <= 0:
raise ValueError("O preço deve ser superior a zero.")
return v
# DTO de Saída: Determina o formato exato que o JSON final de retorno assumirá
class ProdutoSchema(BaseModel):
id: int
nome: str
descricao: Optional[str]
preco: Decimal
estoque: int
categoria_id: int
categoria_nome: str
class Config:
# Permite que o Pydantic leia dados direto de objetos ORM do SQLAlchemy
from_attributes = True
💳 2. Schemas de Pedido e Carrinho
Crie o arquivo em app/schemas/pedido_schema.py:
from datetime import datetime
from decimal import Decimal
from typing import List
from pydantic import BaseModel, Field
# Item unitário recebido no checkout
class ItemPedidoFormSchema(BaseModel):
produto_id: int = Field(..., ge=1)
quantidade: int = Field(..., ge=1, description="Quantidade mínima de compra é 1")
# Checkout completo enviado pelo Vue 3
class PedidoFormSchema(BaseModel):
cliente_id: int = Field(..., ge=1)
itens: List[ItemPedidoFormSchema] = Field(..., min_length=1)
# Item de pedido faturado para retorno JSON
class ItemPedidoSchema(BaseModel):
id: int
produto_id: int
produto_nome: str
quantidade: int
preco_unitario: Decimal
subtotal: Decimal
class Config:
from_attributes = True
# Retorno do Pedido Completo
class PedidoSchema(BaseModel):
id: int
data_pedido: datetime
status: str
cliente_id: int
cliente_nome: str
itens: List[ItemPedidoSchema]
valor_total: Decimal
class Config:
from_attributes = True
🔄 2. Conversores Manuais (Mappers)
Para manter a clareza didática das transições ORM ➔ DTO, criaremos funções utilitárias que mapeiam explicitamente os dados do banco para os schemas de saída do Pydantic. Isso previne loops de serialização recursivos infinitos causados por relacionamentos bidirecionais das tabelas.
Crie o arquivo em app/schemas/mappers.py:
flowchart LR
A[Banco de Dados] -->|SQLAlchemy ORM| B(Classe Produto)
B -->|Mapper.produto_to_schema| C(ProdutoSchema DTO)
C -->|FastAPI Response| D{JSON Frontend}
style B fill:#bbf,stroke:#333,stroke-width:2px
style C fill:#bfb,stroke:#333,stroke-width:2px
from app.models.produto import Produto
from app.models.pedido import Pedido
from app.schemas.produto_schema import ProdutoSchema
from app.schemas.pedido_schema import PedidoSchema, ItemPedidoSchema
class Mapper:
@staticmethod
def produto_to_schema(p: Produto) -> ProdutoSchema:
return ProdutoSchema(
id=p.id,
nome=p.nome,
descricao=p.descricao,
preco=p.preco,
estoque=p.estoque,
categoria_id=p.categoria_id,
categoria_nome=p.categoria.nome
)
@staticmethod
def pedido_to_schema(ped: Pedido) -> PedidoSchema:
itens_dto = []
valor_total = 0
for item in ped.itens:
subtotal = item.preco_unitario * item.quantidade
valor_total += subtotal
itens_dto.append(ItemPedidoSchema(
id=item.id,
produto_id=item.produto_id,
produto_nome=item.produto.nome,
quantidade=item.quantidade,
preco_unitario=item.preco_unitario,
subtotal=subtotal
))
return PedidoSchema(
id=ped.id,
data_pedido=ped.data_pedido,
status=ped.status,
cliente_id=ped.cliente_id,
cliente_nome=ped.cliente.nome,
itens=itens_dto,
valor_total=valor_total
)
🚨 3. Tratamento de Exceções Globalizado
Uma API de engenharia de software não pode retornar erros com código “500 Internal Server Error” expondo erros brutos de banco de dados ou “stack trace” Python para o navegador do cliente.
Passo 1: Definir as Exceções de Negócio
Crie o arquivo app/exceptions.py:
# Erros 404 (Recurso Não Encontrado)
class ResourceNotFoundException(Exception):
def __init__(self, message: str):
self.message = message
# Erros 400 (Violação de Regra de Negócio, ex: Estoque esgotado)
class BusinessException(Exception):
def __init__(self, message: str):
self.message = message
Passo 2: Registrar os Interceptadores Globais
No arquivo central da API, registraremos listeners que capturam estas exceções e as convertem em um padrão JSON homogêneo.
Crie/edite o arquivo app/main.py:
from datetime import datetime
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from app.exceptions import ResourceNotFoundException, BusinessException
app = FastAPI(title="TecLoja 02 API")
# Handler Global para Recurso Não Encontrado (404)
@app.exception_handler(ResourceNotFoundException)
async def handle_not_found(request: Request, exc: ResourceNotFoundException):
return JSONResponse(
status_code=status.HTTP_404_NOT_FOUND,
content={
"timestamp": datetime.utcnow().isoformat(),
"status": status.HTTP_404_NOT_FOUND,
"error": "Resource Not Found",
"message": exc.message,
"path": request.url.path
}
)
# Handler Global para Erros de Negócio (400)
@app.exception_handler(BusinessException)
async def handle_business(request: Request, exc: BusinessException):
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={
"timestamp": datetime.utcnow().isoformat(),
"status": status.HTTP_400_BAD_REQUEST,
"error": "Business Rule Violation",
"message": exc.message,
"path": request.url.path
}
)
# Handler Global para Erros de Validação de Input do Pydantic (422)
from fastapi.exceptions import RequestValidationError
@app.exception_handler(RequestValidationError)
async def handle_validation(request: Request, exc: RequestValidationError):
errors = {err["loc"][-1]: err["msg"] for err in exc.errors()}
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content={
"timestamp": datetime.utcnow().isoformat(),
"status": status.HTTP_422_UNPROCESSABLE_ENTITY,
"error": "Validation Failed",
"message": "Alguns campos inseridos são inválidos.",
"validationErrors": errors,
"path": request.url.path
}
)
✅ Pré-Requisitos deste Módulo
Antes de avançar para a validação de contratos de dados, certifique-se de que:
- A base de dados local
tecloja.dbfoi gerada e povoada via migrations/seeder no Módulo 02. - A pasta
app/schemas/foi criada no seu repositório backend.
🤔 Por que fizemos assim?
- Por que usar Schemas do Pydantic como DTOs? DTOs (Data Transfer Objects) servem para isolar a representação do banco de dados (Entidades do SQLAlchemy) do contrato de comunicação externo da API (JSON). Isso impede que alterações nas tabelas quebrem diretamente o frontend do Vue 3, além de impedir que o usuário envie dados extras maliciosos (ex: injetar IDs ou forçar descontos).
- Por que criar Mappers estáticos explícitos? Embora o Pydantic possua a configuração
from_attributes = Truepara ler objetos do banco, o uso direto em objetos com relacionamentos bidirecionais complexos (ex: Produto tem Categoria e Categoria tem uma lista de Produtos) pode gerar loops infinitos de serialização de JSON. Conversores explícitos como oMappereliminam essa possibilidade e nos permitem calcular campos dinâmicos no código (como osubtotal). - Por que reescrever o handler de erro de validação (
RequestValidationError)? Por padrão, as falhas de validação do Pydantic retornam uma lista de detalhes interna e confusa para o usuário final. Modificar este interceptador nos permite estruturar um dicionário simples de chave-valor (ex:"preco": "O preço deve ser superior a zero."), tornando a exibição de alertas de validação de formulários muito mais simples no Vue.
🔍 Checkpoint
- Criação dos Schemas: Garanta que os arquivos de contrato estão salvos em:
tecloja-backend/app/schemas/produto_schema.pytecloja-backend/app/schemas/pedido_schema.pytecloja-backend/app/schemas/mappers.py
- Interceptadores Ativos: Confirme que o arquivo
app/main.pycontém os blocos@app.exception_handlercom retorno do tipoJSONResponse.
⚠️ Erros Comuns
| Erro | Causa | Solução |
|---|---|---|
AttributeError: 'greenlet.error' object has no attribute '...' |
O Mapper tentou acessar uma propriedade de relacionamento que não foi carregada no banco (ex: p.categoria.nome), disparando erro do Greenlet assíncrono. |
Certifique-se de que a query de busca no banco utilizou carregamento antecipado (Eager Loading) do relacionamento referenciado no Mapper. |
Erros de importação circular em mappers.py |
Importar classes de entidades e schemas de arquivos que já se importam mutuamente. | Centralize importações de schemas específicos no arquivo do Mapper e utilize anotações de string se as classes necessitarem de referências cruzadas. |
Configuração antiga de ORM orm_mode = True gerando Warning |
O Pydantic v2 renomeou as propriedades de suporte a objetos ORM. | Em projetos modernos que utilizam Pydantic v2+, utilize sempre from_attributes = True dentro da subclasse class Config de seus schemas. |
🏁 Conclusão
Excelente! Nosso backend agora possui uma blindagem de dados completa. Toda e qualquer entrada é minuciosamente validada por schemas do Pydantic v2 e os erros gerados no sistema serão capturados, formatados de forma profissional e devolvidos ao Vue de maneira clara e segura.
No Módulo 04, programaremos a Camada de Serviço (Regras de Negócio). Desenvolveremos a orquestração do checkout de pedidos, manipulando as transações assíncronas do SQLAlchemy 2.0 e lidando com integridade referencial física e Rollback transacional de estoque.