SQLの論理関数について学び、データ値に基づいて意思決定を行う方法を習得しました。このチャレンジでは、その知識を活用してpaymentsテーブル内の疑わしい取引を特定します。`amount`が`1000`以上、または`status`が`"failed"`の場合、その取引を**「Suspicious」**としてフラグ付けすることが目標です。これは不正検出やリスク管理においてよくあるシナリオであり、追加の確認が必要な取引を強調することが重要です。

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 test_flag_column_present(self):
        """Check that the result includes a 'flag' column."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        condition = 'flag' in [col.lower() for col in columns]
        _dynamic_test(
            self,
            condition,
            "The result includes a 'flag' column.",
            "The result does not include a 'flag' column."
        )

    def test_amount_1000_or_more_flagged_suspicious(self):
        """Transactions with amount >= 1000 are flagged as 'Suspicious'."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Add a transaction >= 1000 for this test
        self.cursor.execute("INSERT INTO transactions (transaction_id, amount, status, payment_method) VALUES (100, 1200.00, 'completed', 'credit_card') ON CONFLICT DO NOTHING;")
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        flagged = df[(df['transaction_id'] == 100)]
        condition = not flagged.empty and normalize_text(flagged.iloc[0]['flag']) == 'suspicious'
        _dynamic_test(
            self,
            condition,
            "Transactions with amount >= 1000 are flagged as 'Suspicious'.",
            "A transaction with amount >= 1000 was not flagged as 'Suspicious'."
        )

    def test_status_failed_flagged_suspicious(self):
        """Transactions with status = 'failed' are flagged as 'Suspicious'."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        failed_rows = df[df['status'].apply(lambda x: normalize_text(str(x)) == 'failed')]
        suspicious_flags = failed_rows['flag'].apply(lambda x: normalize_text(str(x)) == 'suspicious')
        condition = suspicious_flags.all() and not failed_rows.empty
        _dynamic_test(
            self,
            condition,
            "All transactions with status = 'failed' are flagged as 'Suspicious'.",
            "Not all transactions with status = 'failed' are flagged as 'Suspicious'."
        )

    def test_other_transactions_flagged_normal(self):
        """All other transactions are flagged as 'Normal'."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        df = pd.DataFrame(rows, columns=columns)
        # Exclude rows where amount >= 1000 or status = 'failed'
        mask = (df['amount'] < 1000) & (df['status'].apply(lambda x: normalize_text(str(x)) != 'failed'))
        normal_rows = df[mask]
        if not normal_rows.empty:
            all_normal = normal_rows['flag'].apply(lambda x: normalize_text(str(x)) == 'normal').all()
        else:
            all_normal = True
        _dynamic_test(
            self,
            all_normal,
            "All other transactions are flagged as 'Normal'.",
            "Not all other transactions are flagged as 'Normal'."
        )

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

組み込みSQL関数を活用して、文字列、日付、数値、論理値、NULL、JSONデータを操作する方法を習得します。本モジュールでは、実践的なアプローチでSQL関数の理解と応用を深め、現実のシナリオにおけるデータクエリおよび変換スキルを向上させます。

データベース内のテキストデータを操作および分析するための、SQLの強力な文字列関数を学習します。

チャレンジ：不審な取引のフラグ付け

解答