Esaminiamo insieme la creazione di una transazione basata sulla tabella `BankAccounts` descritta nel capitolo precedente.

Eseguiremo una semplice operazione finanziaria: il trasferimento di fondi da un conto a un altro. Garantire che entrambi i passaggi avvengano come un'unica unità logica rappresenta un esempio concreto di transazione reale.

Lavoreremo con i numeri di conto `101` e `789` e il saldo iniziale sarà il seguente:

import unittest
import psycopg2
import pandas as pd

class TestPostgresQuery(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.connection = psycopg2.connect(
            host="localhost",
            user="agent",
            password="qwerty",
            dbname="testdb",
            sslmode="disable"
        )
        cls.cursor = cls.connection.cursor()

    def test_balances_after_transaction(self):
        """Check if balances for accounts 101 and 789 are updated correctly after transferring 500 from 101 to 789."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("UPDATE BankAccounts SET balance = 4500 WHERE account_number = 101;")
        self.cursor.execute("UPDATE BankAccounts SET balance = 3500 WHERE account_number = 789;")
        self.connection.commit()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)
        result_df = result_df.set_index('account_number')
        expected_balances = {101: 4000.00, 789: 4000.00}
        correct = True
        for acc, expected in expected_balances.items():
            if acc not in result_df.index or float(result_df.loc[acc, 'balance']) != expected:
                correct = False
        _dynamic_test(
            self,
            correct,
            "Balances are updated correctly after transaction.",
            f"Balances are not updated correctly.\\nExpected: {expected_balances}\\nGot: {result_df['balance'].to_dict()}"
        )

    def test_transaction_committed(self):
        """Check that the transaction is committed and balances persist after reconnect."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Reset balances
        self.cursor.execute("UPDATE BankAccounts SET balance = 4500 WHERE account_number = 101;")
        self.cursor.execute("UPDATE BankAccounts SET balance = 3500 WHERE account_number = 789;")
        self.connection.commit()
        self.cursor.execute(query)
        self.connection.commit()
        # Reconnect to check persistence
        self.cursor.close()
        self.connection.close()
        connection2 = psycopg2.connect(
            host="localhost",
            user="agent",
            password="qwerty",
            dbname="testdb",
            sslmode="disable"
        )
        cursor2 = connection2.cursor()
        cursor2.execute("SELECT account_number, balance FROM BankAccounts WHERE account_number IN (101, 789);")
        rows = cursor2.fetchall()
        result_df = pd.DataFrame(rows, columns=['account_number', 'balance']).set_index('account_number')
        expected_balances = {101: 4000.00, 789: 4000.00}
        correct = True
        for acc, expected in expected_balances.items():
            if acc not in result_df.index or float(result_df.loc[acc, 'balance']) != expected:
                correct = False
        _dynamic_test(
            self,
            correct,
            "Transaction was committed and balances persisted.",
            f"Transaction was not committed or balances did not persist.\\nExpected: {expected_balances}\\nGot: {result_df['balance'].to_dict()}"
        )
        cursor2.close()
        connection2.close()
def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\\s{2,}", " ", text)
    text = re.sub(r"\\s*([,:?])\\s*", r"\\1 ", text)
    return text.strip()

if __name__ == "__main__":
    unittest.main()


test_solution.py

Inizia un entusiasmante percorso attraverso SQL con Tecniche Avanzate in SQL. Approfondisci le proprietà fondamentali ACID che garantiscono l'integrità dei dati nelle transazioni di database. Scopri metodi avanzati per ottimizzare le query e migliorare le prestazioni del database. Esplora funzionalità speciali di SQL come funzioni di finestra, trigger, Data Control Language (DCL) e database NoSQL. Questo ti permetterà di gestire la moderna amministrazione dei database con sicurezza ed efficacia.

Ora esploriamo i principi fondamentali di ACID (Atomicità, Consistenza, Isolamento, Durabilità) nella gestione dei database. Questi principi garantiscono transazioni affidabili assicurando operazioni complete, mantenendo i dati consistenti, isolando le transazioni concorrenti e garantendo la memorizzazione permanente dei dati.

Esplora come migliorare l'efficienza delle query SQL tramite l'ottimizzazione degli indici. Approfondisci l'importanza degli indici nelle prestazioni dei database, imparando a progettarli, implementarli e utilizzarli in modo efficace. Scopri strategie per la scelta degli indici appropriati, comprendendo il loro impatto sui piani di esecuzione delle query e ottimizzando le prestazioni delle query attraverso la selezione e l'utilizzo degli indici.

Questa sezione esplora argomenti fondamentali come trigger, funzioni finestra e DCL (Data Control Language), offrendo approfondimenti completi sulle strategie di implementazione e ottimizzazione. Inoltre, analizza il panorama dei DBMS NoSQL, fornendo una comprensione approfondita della loro architettura, dei vantaggi e delle applicazioni pratiche.

Sfida: Creazione di Transazioni

Soluzione

Sfida: Creazione di Transazioni

Soluzione