This short video introduces the challenge of safely transferring funds between two bank accounts using a SQL transaction. It demonstrates how to structure the transaction so that both the deduction and addition of funds occur together, and shows how to ensure the entire transaction is rolled back if the source account does not have sufficient funds.

When transferring funds between two accounts in a banking system, you must ensure that both the deduction from the source account and the addition to the destination account happen together as a single, indivisible operation. This is where the concept of a transaction comes in: a transaction groups multiple SQL statements so that they either all succeed or none take effect, maintaining atomicity.

To transfer funds, you will work with the `bank_accounts` table, which contains columns for `account_id`, `owner_name`, and `balance`. The transaction should:

- Start with a `BEGIN` statement to initiate the transaction;
- Check that the source account has enough funds for the transfer;
- If there are sufficient funds, deduct the amount from the source account and add it to the destination account;
- If the source account lacks sufficient funds, the transaction should be rolled back, leaving all balances unchanged;
- Commit the transaction only if all steps succeed.

This approach ensures that no partial updates occur, and the system remains reliable even in the event of errors.

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 setUp(self):
        # Reset state before each test
        self.connection.rollback()
        self.cursor.execute("UPDATE bank_accounts SET balance = 5000.00 WHERE account_id = 1;")
        self.cursor.execute("UPDATE bank_accounts SET balance = 3200.50 WHERE account_id = 2;")
        self.cursor.execute("DELETE FROM transfers WHERE from_account = 1 AND to_account = 2 AND amount = 1000.00;")
        self.connection.commit()

    def test_deducts_from_source_on_success(self):
        """Deducts amount from source account only if sufficient funds exist."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Run user query
        self.cursor.execute(query)
        # Check new balance
        self.cursor.execute("SELECT balance FROM bank_accounts WHERE account_id = 1;")
        balance = float(self.cursor.fetchone()[0])
        _dynamic_test(
            self,
            abs(balance - 4000.00) < 0.01,
            "Source account balance correctly deducted.",
            f"Source account balance not correctly deducted. Expected 4000.00, got {balance}."
        )

    def test_adds_to_destination_on_success(self):
        """Adds amount to destination account only if source has sufficient funds."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        self.cursor.execute("SELECT balance FROM bank_accounts WHERE account_id = 2;")
        balance = float(self.cursor.fetchone()[0])
        _dynamic_test(
            self,
            abs(balance - 4200.50) < 0.01,
            "Destination account balance correctly increased.",
            f"Destination account balance not correctly increased. Expected 4200.50, got {balance}."
        )

    def test_transfer_recorded_on_success(self):
        """Records the transfer in transfers table only if sufficient funds exist."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        self.cursor.execute("SELECT COUNT(*) FROM transfers WHERE from_account = 1 AND to_account = 2 AND amount = 1000.00;")
        count = self.cursor.fetchone()[0]
        _dynamic_test(
            self,
            count >= 1,
            "Transfer record inserted successfully.",
            "Transfer record was not inserted."
        )

    def test_rollback_on_insufficient_funds(self):
        """Rolls back all changes if source account does not have enough funds."""
        # Set source balance to less than transfer_amount
        self.cursor.execute("UPDATE bank_accounts SET balance = 500.00 WHERE account_id = 1;")
        self.connection.commit()
        with open("query.sql", "r") as file:
            query = file.read()
        try:
            self.cursor.execute(query)
        except Exception:
            self.connection.rollback()
        # Check balances and transfer records
        self.cursor.execute("SELECT balance FROM bank_accounts WHERE account_id = 1;")
        balance1 = float(self.cursor.fetchone()[0])
        self.cursor.execute("SELECT balance FROM bank_accounts WHERE account_id = 2;")
        balance2 = float(self.cursor.fetchone()[0])
        self.cursor.execute("SELECT COUNT(*) FROM transfers WHERE from_account = 1 AND to_account = 2 AND amount = 1000.00;")
        count = self.cursor.fetchone()[0]
        all_unchanged = abs(balance1 - 500.00) < 0.01 and abs(balance2 - 3200.50) < 0.01 and count == 0
        _dynamic_test(
            self,
            all_unchanged,
            "Transaction rolled back and no changes made on insufficient funds.",
            f"Balances or transfer records changed on insufficient funds. (Balances: {balance1}, {balance2}; Transfers: {count})"
        )

    def test_no_partial_update_on_failure(self):
        """Ensures no partial updates occur if transaction fails."""
        # Set up to cause failure
        self.cursor.execute("UPDATE bank_accounts SET balance = 700.00 WHERE account_id = 1;")
        self.connection.commit()
        with open("query.sql", "r") as file:
            query = file.read()
        try:
            self.cursor.execute(query)
        except Exception:
            self.connection.rollback()
        # Check all effects
        self.cursor.execute("SELECT balance FROM bank_accounts WHERE account_id = 1;")
        balance1 = float(self.cursor.fetchone()[0])
        self.cursor.execute("SELECT balance FROM bank_accounts WHERE account_id = 2;")
        balance2 = float(self.cursor.fetchone()[0])
        self.cursor.execute("SELECT COUNT(*) FROM transfers WHERE from_account = 1 AND to_account = 2 AND amount = 1000.00;")
        count = self.cursor.fetchone()[0]
        no_partial = abs(balance1 - 700.00) < 0.01 and abs(balance2 - 3200.50) < 0.01 and count == 0
        _dynamic_test(
            self,
            no_partial,
            "No partial updates made when transaction fails.",
            f"Partial updates occurred on transaction failure. (Balances: {balance1}, {balance2}; Transfers: {count})"
        )

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

Learn about the important ACID properties that help keep data accurate in database transactions. Discover advanced ways to optimize your queries to improve database performance. 

Master advanced SQL concepts including ACID properties, query optimization, indexing, triggers, window functions, and data control. This course is designed for learners with a solid SQL foundation who want to deepen their understanding and practical skills in optimizing queries and leveraging advanced database features.

Challenge: Transaction Creation

Solution