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.

銀行システムで2つの口座間で資金を移動する際には、送金元口座からの引き落としと送金先口座への加算が、単一かつ不可分な操作として同時に行われる必要があります。ここでトランザクションの概念が重要となります。トランザクションは複数のSQL文をまとめて実行し、すべてが成功するか、または一切反映されないようにすることで、原子性を維持します。

資金移動には、`bank_accounts` テーブルを使用します。このテーブルには `account_id`、`owner_name`、`balance` のカラムがあります。トランザクションは以下の手順で実施します：

- トランザクション開始のために `BEGIN` 文を実行
- 送金元口座に十分な残高があるかを確認
- 十分な残高がある場合、送金元口座から金額を差し引き、送金先口座に加算
- 送金元口座の残高が不足している場合、トランザクションをロールバックし、すべての残高を変更前の状態に戻す
- すべての手順が成功した場合のみ、トランザクションをコミット

この方法により、部分的な更新が発生せず、エラーが発生した場合でもシステムの信頼性が保たれます。

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

データベーストランザクションにおけるデータの正確性を維持するための重要なACID特性について学びます。データベースのパフォーマンスを向上させるための高度なクエリ最適化手法を紹介します。

ACID特性、クエリ最適化、インデックス、トリガー、ウィンドウ関数、データ制御など、SQLの高度な概念を習得します。本コースは、SQLの基礎をしっかりと身につけている学習者が、クエリの最適化や高度なデータベース機能の活用に関する理解と実践的スキルをさらに深めることを目的としています。

チャレンジ：トランザクション作成

解答