**Video: Challenge Introduction — Specifying Transaction Isolation Level in SQL**

This short video introduces the challenge of working with transaction isolation levels in SQL. You will see how to set the isolation level for a transaction, and learn why the Repeatable Read isolation level is important for preventing phantom reads when transferring funds between accounts.

In this challenge, you will practice setting the isolation level for a transaction to ensure data consistency during a funds transfer. Using the `bank_accounts` and `transfers` tables, your goal is to write a SQL transaction that sets the isolation level to **Repeatable Read** before performing a transfer of funds between two accounts. The transaction should ensure that no phantom reads can occur, meaning that the set of rows read at the beginning of the transaction remains unchanged for the duration of the transaction—even if other concurrent transactions attempt to insert or modify relevant data.

You will use the following steps: set the isolation level, begin the transaction, select the current balances, update the balances, insert a record into the `transfers` table, and commit the transaction. The transaction must be robust against phantom reads, as required by the Repeatable Read isolation level.

import unittest
import psycopg2
import pandas as pd
import re

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 balances and transfers table before each test
        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_isolation_level_repeatable_read(self):
        """Check if isolation level is set to Repeatable Read."""
        with open("query.sql", "r") as file:
            query = file.read()
        first_line = query.strip().split(";", 1)[0]
        condition = normalize_text(first_line) == "set transaction isolation level repeatable read"
        _dynamic_test(
            self,
            condition,
            "Isolation level is set to Repeatable Read.",
            "Isolation level is not set to Repeatable Read at the start of the transaction."
        )

    def test_balance_decreased_account_1(self):
        """Check if account 1's balance decreased by 1000.00."""
        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 = 1;")
        balance = float(self.cursor.fetchone()[0])
        condition = abs(balance - 4000.00) < 0.01
        _dynamic_test(
            self,
            condition,
            "Account 1's balance decreased by 1000.00.",
            f"Account 1's balance is {balance}, expected 4000.00."
        )

    def test_balance_increased_account_2(self):
        """Check if account 2's balance increased by 1000.00."""
        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])
        condition = abs(balance - 4200.50) < 0.01
        _dynamic_test(
            self,
            condition,
            "Account 2's balance increased by 1000.00.",
            f"Account 2's balance is {balance}, expected 4200.50."
        )

    def test_transfer_record_inserted(self):
        """Check if a new transfer record is inserted for the transaction."""
        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]
        condition = count >= 1
        _dynamic_test(
            self,
            condition,
            "Transfer record inserted into transfers table.",
            "No transfer record found for the transaction."
        )

    def test_all_within_single_transaction(self):
        """Check if all operations are performed within a single transaction block."""
        with open("query.sql", "r") as file:
            query = file.read()
        normalized = normalize_text(query)
        condition = (
            "begin;" in normalized and
            "commit;" in normalized and
            normalized.index("begin;") < normalized.index("commit;")
        )
        _dynamic_test(
            self,
            condition,
            "All operations are performed within a single transaction block.",
            "Operations are not performed within a single transaction block."
        )

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: Specifying Isolation Level for Transaction

Solution