📚 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:


🤔 Por que fizemos assim?


🔍 Checkpoint

  1. Criação dos Schemas: Garanta que os arquivos de contrato estão salvos em:
    • tecloja-backend/app/schemas/produto_schema.py
    • tecloja-backend/app/schemas/pedido_schema.py
    • tecloja-backend/app/schemas/mappers.py
  2. Interceptadores Ativos: Confirme que o arquivo app/main.py contém os blocos @app.exception_handler com retorno do tipo JSONResponse.

⚠️ 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.


Voltar para o Sumário