Analicemos cómo crear una transacción nosotros mismos utilizando la tabla `BankAccounts` descrita en el capítulo anterior.

Realizaremos una transacción financiera sencilla: transferir fondos de una cuenta a otra. Garantizar que ambos pasos ocurran como una sola unidad lógica ejemplifica una transacción en la vida real.

Trabajaremos con los números de cuenta `101` y `789` y el saldo inicial será el siguiente:

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

Comience un emocionante recorrido por SQL con Técnicas Avanzadas en SQL. Aprenda sobre las importantes propiedades ACID que ayudan a mantener la precisión de los datos en las transacciones de bases de datos. Descubra métodos avanzados para optimizar sus consultas y mejorar el rendimiento de la base de datos. Explore características especiales de SQL como funciones de ventana, triggers, Lenguaje de Control de Datos (DCL) y bases de datos NoSQL. Esto le permitirá gestionar bases de datos modernas con confianza y eficacia.

Ahora exploramos los principios básicos de ACID (Atomicidad, Consistencia, Aislamiento, Durabilidad) en la gestión de bases de datos. Estos principios garantizan transacciones fiables al asegurar operaciones completas, mantener la consistencia de los datos, aislar transacciones concurrentes y garantizar el almacenamiento permanente de la información.

Explore cómo mejorar la eficiencia de las consultas SQL mediante la optimización de índices. Profundice en la importancia de los índices en el rendimiento de la base de datos, aprendiendo a diseñarlos, implementarlos y utilizarlos de manera efectiva. Descubra estrategias para seleccionar los índices apropiados, comprender su impacto en los planes de ejecución de consultas y optimizar el rendimiento de las consultas a través de la selección y utilización de índices.

Esta sección explora temas fundamentales como disparadores, funciones de ventana y DCL (Data Control Language), ofreciendo perspectivas integrales sobre su implementación y estrategias de optimización. Además, aborda el panorama de los sistemas gestores de bases de datos NoSQL, proporcionando una comprensión exhaustiva de su arquitectura, ventajas y aplicaciones prácticas.

Desafío: Creación de Transacciones

Solución

Desafío: Creación de Transacciones

Solución