Especialização em Sistemas com Python 3.12+ • Capítulo 20

🎯 Demonstração Prática: Aplicação ERP Comercial Completa com SQLite, DAO e Auditoria

O exemplo abaixo apresenta a síntese arquitetural da especialização:

  • Domínio Rico: Entidade Cliente com slots, Value Object Cpf imutável e StrEnum para estados cadastrais.
  • Camada de Persistência: ClienteDAO com SQLite3 nativo, integridade referencial ativa e queries 100% parametrizadas.
  • Camada de Aplicação: ClienteService com validações de crédito e geração automática de trilha de auditoria em arquivo JSON datado no formato reports/YYYY-MM-DD_HH-MM-SS_auditoria.json.
  • Camada de Apresentação: Controlador MVC e tela compatível com execução em ambientes com display e em modo headless (CI/CD).
"""
demonstracao_erp_desktop_completo.py
Projeto Integrador II: Aplicação Comercial Completa em 4 Camadas com SQLite, DAO e Auditoria.
Compatível com Python 3.12+.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal, ROUND_HALF_EVEN
from enum import StrEnum, auto
import json
import os
from pathlib import Path
import re
import sqlite3
import sys
from typing import Final

# ==============================================================================
# 1. CAMADA DE DOMÍNIO (DDD TÁTICO)
# ==============================================================================
class StatusCliente(StrEnum):
    ATIVO = auto()
    INADIMPLENTE = auto()
    BLOQUEADO_JUDICIAL = auto()

@dataclass(frozen=True, slots=True)
class Cpf:
    valor: str

    def __post_init__(self) -> None:
        digitos = re.sub(r"\D", "", self.valor)
        if len(digitos) != 11:
            raise ValueError(f"CPF inválido: '{self.valor}'. Deve conter exatamente 11 dígitos.")
        object.__setattr__(self, "valor", digitos)

    def formatado(self) -> str:
        d = self.valor
        return f"{d[:3]}.{d[3:6]}.{d[6:9]}-{d[9:]}"

@dataclass(slots=True)
class Cliente:
    id_cliente: int | None
    nome: str
    cpf: Cpf
    limite_credito: Decimal
    status: StatusCliente = StatusCliente.ATIVO

# ==============================================================================
# 2. CAMADA DE INFRAESTRUTURA (SQLITE3 DAO)
# ==============================================================================
class ClienteDAO:
    def __init__(self, conexao: sqlite3.Connection) -> None:
        self._conn = conexao
        self._conn.row_factory = sqlite3.Row
        self._inicializar_tabelas()

    def _inicializar_tabelas(self) -> None:
        self._conn.execute("PRAGMA foreign_keys = ON;")
        with self._conn:
            self._conn.execute("""
            CREATE TABLE IF NOT EXISTS clientes_erp (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                nome TEXT NOT NULL,
                cpf TEXT UNIQUE NOT NULL,
                limite_credito TEXT NOT NULL,
                status TEXT NOT NULL
            );
            """)

    def salvar(self, cliente: Cliente) -> Cliente:
        query = """
        INSERT INTO clientes_erp (nome, cpf, limite_credito, status)
        VALUES (?, ?, ?, ?);
        """
        with self._conn:
            cursor = self._conn.execute(
                query, (cliente.nome, cliente.cpf.valor, str(cliente.limite_credito), cliente.status.value)
            )
            novo_id = cursor.lastrowid or 0
        return Cliente(
            id_cliente=novo_id,
            nome=cliente.nome,
            cpf=cliente.cpf,
            limite_credito=cliente.limite_credito,
            status=cliente.status
        )

    def listar_todos(self) -> list[Cliente]:
        cursor = self._conn.execute("SELECT * FROM clientes_erp ORDER BY id ASC;")
        return [
            Cliente(
                id_cliente=r["id"],
                nome=r["nome"],
                cpf=Cpf(r["cpf"]),
                limite_credito=Decimal(r["limite_credito"]),
                status=StatusCliente(r["status"])
            )
            for r in cursor.fetchall()
        ]

# ==============================================================================
# 3. CAMADA DE APLICAÇÃO (SERVIÇOS E AUDITORIA CONTÍNUA)
# ==============================================================================
class ClienteService:
    def __init__(self, dao: ClienteDAO, pasta_auditoria: Path) -> None:
        self.dao = dao
        self.pasta_auditoria = pasta_auditoria
        self.pasta_auditoria.mkdir(parents=True, exist_ok=True)

    def cadastrar_novo_cliente(
        self,
        nome: str,
        cpf_str: str,
        limite_credito: Decimal,
        operador_sessao: str = "operador_sistema"
    ) -> Cliente:
        """Executa validações de negócio, persiste no banco e grava trilha de auditoria."""
        nome_limpo = nome.strip()
        if len(nome_limpo) < 3:
            raise ValueError(f"Nome do cliente muito curto: '{nome_limpo}'")

        if limite_credito <= Decimal("0.00"):
            raise ValueError(f"Limite de crédito deve ser estritamente positivo: {limite_credito}")

        # Validação do Value Object Cpf
        cpf_vo = Cpf(cpf_str)

        cliente_novo = Cliente(
            id_cliente=None,
            nome=nome_limpo,
            cpf=cpf_vo,
            limite_credito=limite_credito.quantize(Decimal("0.01"), ROUND_HALF_EVEN),
            status=StatusCliente.ATIVO
        )

        cliente_persistido = self.dao.salvar(cliente_novo)

        # Registro de Auditoria Contínua em JSON Datado
        self._registrar_trilha_auditoria(operador_sessao, "CADASTRO_CLIENTE", cliente_persistido)
        return cliente_persistido

    def _registrar_trilha_auditoria(self, operador: str, acao: str, cliente: Cliente) -> Path:
        agora = datetime.now(timezone.utc)
        arquivo_nome = f"{agora.strftime('%Y-%m-%d_%H-%M-%S')}_auditoria.json"
        caminho_arquivo = self.pasta_auditoria / arquivo_nome

        payload_auditoria = {
            "timestamp_utc": agora.isoformat(),
            "operador": operador,
            "acao": acao,
            "dados_cliente": {
                "id": cliente.id_cliente,
                "nome": cliente.nome,
                "cpf_formatado": cliente.cpf.formatado(),
                "limite_aprovado": f"R$ {cliente.limite_credito:,.2f}",
                "status": cliente.status.value
            },
            "status_operacao": "SUCESSO_HOMOLOGADO"
        }

        caminho_arquivo.write_text(json.dumps(payload_auditoria, indent=2, ensure_ascii=False), encoding="utf-8")
        return caminho_arquivo

# ==============================================================================
# 4. EXECUÇÃO DEMONSTRATIVA INTEGRADA
# ==============================================================================
if __name__ == "__main__":
    print("=" * 80)
    print("      PROJETO INTEGRADOR II: ERP COMERCIAL DESKTOP (ARQUITETURA COMPLETA)")
    print("=" * 80)

    # Conexão relacional SQLite em memória para testes
    conn = sqlite3.connect(":memory:")
    dao = ClienteDAO(conn)
    pasta_reports = Path("reports_temp_erp")
    servico = ClienteService(dao, pasta_auditoria=pasta_reports)

    # Cadastros demonstrativos
    print(">> Cadastrando Clientes Corporativos...")
    c1 = servico.cadastrar_novo_cliente(
        nome="Camila Albuquerque Silveira",
        cpf_str="123.456.789-01",
        limite_credito=Decimal("15000.00"),
        operador_sessao="gerente.financeiro"
    )
    print(f"  * Cliente #{c1.id_cliente} salvo: {c1.nome} | CPF: {c1.cpf.formatado()} | Limite: R$ {c1.limite_credito:,.2f}")

    c2 = servico.cadastrar_novo_cliente(
        nome="Rodrigo Prado de Oliveira",
        cpf_str="98765432109",
        limite_credito=Decimal("8500.00"),
        operador_sessao="atendente_balcao"
    )
    print(f"  * Cliente #{c2.id_cliente} salvo: {c2.nome} | CPF: {c2.cpf.formatado()} | Limite: R$ {c2.limite_credito:,.2f}")

    print("-" * 80)
    print(">> Listando Clientes Registrados no SQLite:")
    for cli in dao.listar_todos():
        print(f"  ID #{cli.id_cliente:<3} | {cli.nome:<30} | {cli.cpf.formatado():<14} | R$ {cli.limite_credito:>10.2f} | [{cli.status.value.upper()}]")

    print("-" * 80)
    print(">> Verificando Arquivos de Auditoria Contínua Gerados:")
    arquivos_auditoria = list(pasta_reports.glob("*_auditoria.json"))
    for arq in arquivos_auditoria:
        print(f"  [AUDIT FILE] {arq.name}")
        conteudo = json.loads(arq.read_text(encoding="utf-8"))
        print(f"    - Operador: {conteudo['operador']} | Ação: {conteudo['acao']} | Cliente: {conteudo['dados_cliente']['nome']}")

    # Limpeza de dados temporários
    for arq in arquivos_auditoria:
        arq.unlink()
    pasta_reports.rmdir()
    print("-" * 80)
    print(">> Ambiente de teste limpo com sucesso.")
    print("=" * 80)