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

🎯 Demonstração Prática: Transações ACID e Padrão DAO com SQLite3

O exemplo abaixo implementa uma camada completa de persistência relacional utilizando o módulo nativo sqlite3:

  • Configuração de pragmas mandatórios (PRAGMA foreign_keys = ON e sqlite3.Row).
  • Implementação do padrão Data Access Object (DAO) com queries 100% parametrizadas.
  • Transferência monetária transacional atômica demonstrando garantias ACID (with conn: gerenciando commit e rollback em falhas de saldo).
"""
demonstracao_sqlite_e_dao.py
Camada DAO Relacional com SQLite3 Nativo, Pragmas de Engenharia e Transações ACID.
Compatível com Python 3.12+.
"""

from __future__ import annotations

from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_EVEN
import sqlite3
from typing import Final

# ==============================================================================
# 1. MODELOS DE DOMÍNIO
# ==============================================================================
@dataclass(slots=True)
class ContaCorrente:
    id_conta: int | None
    titular: str
    saldo: Decimal
    bloqueada: bool = False

# ==============================================================================
# 2. DATA ACCESS OBJECT (DAO) COM PRAGMAS E TRANSAÇÕES
# ==============================================================================
class ContaCorrenteDAO:
    """DAO responsável pelo isolamento de comandos SQL e mapeamento de registros."""
    def __init__(self, conexao: sqlite3.Connection) -> None:
        self._conn: sqlite3.Connection = conexao
        self._configurar_ambiente_db()

    def _configurar_ambiente_db(self) -> None:
        """Habilita integridade referencial e row_factory dinâmico."""
        self._conn.row_factory = sqlite3.Row
        self._conn.execute("PRAGMA foreign_keys = ON;")
        self._conn.execute("PRAGMA journal_mode = WAL;")

        with self._conn:
            self._conn.execute("""
            CREATE TABLE IF NOT EXISTS contas (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                titular TEXT NOT NULL,
                saldo TEXT NOT NULL,
                bloqueada INTEGER DEFAULT 0 CHECK (bloqueada IN (0, 1))
            );
            """)

    def criar_conta(self, titular: str, saldo_inicial: Decimal) -> ContaCorrente:
        query = "INSERT INTO contas (titular, saldo, bloqueada) VALUES (?, ?, 0);"
        with self._conn:
            cursor = self._conn.execute(query, (titular, str(saldo_inicial)))
            novo_id = cursor.lastrowid or 0
        return ContaCorrente(id_conta=novo_id, titular=titular, saldo=saldo_inicial)

    def buscar_por_id(self, id_conta: int) -> ContaCorrente | None:
        cursor = self._conn.execute("SELECT * FROM contas WHERE id = ?;", (id_conta,))
        linha = cursor.fetchone()
        if not linha:
            return None
        return ContaCorrente(
            id_conta=linha["id"],
            titular=linha["titular"],
            saldo=Decimal(linha["saldo"]),
            bloqueada=bool(linha["bloqueada"])
        )

    def transferir_fundos_atomico(
        self,
        id_origem: int,
        id_destino: int,
        valor: Decimal
    ) -> bool:
        """
        Executa transferência atômica com garantias ACID completas.
        Se qualquer débito ou crédito falhar, o bloco emite ROLLBACK incondicional.
        """
        if valor <= Decimal("0.00"):
            raise ValueError(f"Valor de transferência inválido: {valor}")

        # Bloco transacional com commit ou rollback automático
        with self._conn:
            # 1. Validação da conta de origem
            cursor_origem = self._conn.execute("SELECT saldo, bloqueada FROM contas WHERE id = ?;", (id_origem,))
            row_origem = cursor_origem.fetchone()
            if not row_origem or bool(row_origem["bloqueada"]):
                raise RuntimeError("Conta de origem inexistente ou bloqueada.")

            saldo_origem = Decimal(row_origem["saldo"])
            if saldo_origem < valor:
                raise ValueError(f"Saldo insuficiente na conta {id_origem}: R$ {saldo_origem:,.2f}")

            # 2. Validação da conta de destino
            cursor_dest = self._conn.execute("SELECT saldo, bloqueada FROM contas WHERE id = ?;", (id_destino,))
            row_dest = cursor_dest.fetchone()
            if not row_dest or bool(row_dest["bloqueada"]):
                raise RuntimeError("Conta de destino inexistente ou bloqueada.")

            saldo_dest = Decimal(row_dest["saldo"])

            # 3. Operações de Débito e Crédito atômicas
            novo_saldo_origem = (saldo_origem - valor).quantize(Decimal("0.01"), ROUND_HALF_EVEN)
            novo_saldo_dest = (saldo_dest + valor).quantize(Decimal("0.01"), ROUND_HALF_EVEN)

            self._conn.execute("UPDATE contas SET saldo = ? WHERE id = ?;", (str(novo_saldo_origem), id_origem))
            self._conn.execute("UPDATE contas SET saldo = ? WHERE id = ?;", (str(novo_saldo_dest), id_destino))

        return True

# ==============================================================================
# 3. EXECUÇÃO DEMONSTRATIVA EM BANCO IN-MEMORY
# ==============================================================================
if __name__ == "__main__":
    print("=" * 80)
    print("      PERSISTÊNCIA RELACIONAL E TRANSAÇÕES ACID COM SQLITE3 NATIVO")
    print("=" * 80)

    # Conexão em memória isolada para validação ágil
    conexao_banco = sqlite3.connect(":memory:")
    dao = ContaCorrenteDAO(conexao_banco)

    # 1. Cadastro inicial de contas
    c1 = dao.criar_conta("Mariana Albuquerque", Decimal("2500.00"))
    c2 = dao.criar_conta("Roberto Mendonça", Decimal("350.00"))

    print(f"Conta Origem (#1) : {c1.titular} | Saldo: R$ {c1.saldo:>10.2f}")
    print(f"Conta Destino (#2): {c2.titular} | Saldo: R$ {c2.saldo:>10.2f}")
    print("-" * 80)

    # 2. Transferência Válida
    print(">> Executando transferência atômica de R$ 800,00 da Conta #1 para #2...")
    sucesso = dao.transferir_fundos_atomico(c1.id_conta, c2.id_conta, Decimal("800.00"))
    print(f"Operação Concluída com Sucesso? : {sucesso}")

    c1_atual = dao.buscar_por_id(1)
    c2_atual = dao.buscar_por_id(2)
    print(f"Novo Saldo Conta #1: R$ {c1_atual.saldo:>10.2f}")
    print(f"Novo Saldo Conta #2: R$ {c2_atual.saldo:>10.2f}")
    print("-" * 80)

    # 3. Teste de Falha e Rollback Automático
    print(">> Tentando transferir R$ 5.000,00 da Conta #1 (Valor acima do saldo disponível)...")
    try:
        dao.transferir_fundos_atomico(c1.id_conta, c2.id_conta, Decimal("5000.00"))
    except ValueError as err_transacao:
        print(f"  [ROLLBACK DISPARADO] {err_transacao}")

    c1_pos_erro = dao.buscar_por_id(1)
    c2_pos_erro = dao.buscar_por_id(2)
    print("\nAuditoria pós-tentativa frustrada (Saldos Intactos):")
    print(f"  * Saldo Conta #1 (Preservado): R$ {c1_pos_erro.saldo:>10.2f}")
    print(f"  * Saldo Conta #2 (Preservado): R$ {c2_pos_erro.saldo:>10.2f}")
    print("=" * 80)