After exploring the basics of logical functions and their application in SQL, you are ready to take on a practical challenge that puts your understanding to the test. In real-world scenarios, it is common to categorize transactions by their amount to quickly identify high-value, medium-value, or low-value activities. This kind of classification is particularly useful for business analytics, fraud detection, and reporting.

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_all_rows_returned(self):
        """Check that all rows from transactions are returned."""
        with open("query.sql", "r") as file:
            query = file.read()

        self.cursor.execute("SELECT COUNT(*) FROM transactions;")
        expected_count = self.cursor.fetchone()[0]
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        _dynamic_test(
            self,
            len(rows) == expected_count,
            f"All {expected_count} rows are returned.",
            f"Expected {expected_count} rows, got {len(rows)}."
        )

    def test_amount_category_column_exists(self):
        """Check that amount_category column exists in result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        _dynamic_test(
            self,
            "amount_category" in columns,
            "amount_category column is present.",
            "amount_category column is missing."
        )

    def test_high_category(self):
        """Transactions with amount >= 300 are labeled 'High'."""
        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)
        high_df = df[df["amount"] >= 300]
        condition = all(high_df["amount_category"] == "High")
        _dynamic_test(
            self,
            condition,
            "All transactions with amount >= 300 are labeled 'High'.",
            "Some transactions with amount >= 300 are not labeled 'High'."
        )

    def test_medium_category(self):
        """Transactions with amount >= 100 and < 300 are labeled 'Medium'."""
        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)
        medium_df = df[(df["amount"] >= 100) & (df["amount"] < 300)]
        condition = all(medium_df["amount_category"] == "Medium")
        _dynamic_test(
            self,
            condition,
            "All transactions with amount >= 100 and < 300 are labeled 'Medium'.",
            "Some transactions with amount >= 100 and < 300 are not labeled 'Medium'."
        )

    def test_low_category(self):
        """Transactions with amount < 100 are labeled 'Low'."""
        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)
        low_df = df[df["amount"] < 100]
        condition = all(low_df["amount_category"] == "Low")
        _dynamic_test(
            self,
            condition,
            "All transactions with amount < 100 are labeled 'Low'.",
            "Some transactions with amount < 100 are not labeled 'Low'."
        )

    def test_all_transactions_present(self):
        """Check that all transaction_ids are present in result."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        result_rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(result_rows, columns=columns)
        self.cursor.execute("SELECT transaction_id FROM transactions;")
        expected_ids = set([row[0] for row in self.cursor.fetchall()])
        returned_ids = set(result_df["transaction_id"])
        condition = expected_ids == returned_ids
        _dynamic_test(
            self,
            condition,
            "All transaction_ids are present in result.",
            f"Missing transaction_ids: {expected_ids - returned_ids}"
        )

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

Master the use of built-in SQL functions to manipulate strings, dates, numbers, logical values, nulls, and JSON data. This course provides a hands-on approach to understanding and applying SQL functions in real-world scenarios, enhancing your data querying and transformation skills.

Explore SQL's powerful string functions for manipulating and analyzing textual data in databases.

Master SQL's date and time functions to work with temporal data effectively.

Learn to use SQL's numeric functions for calculations and data analysis.

Harness SQL's logical functions to control query flow and make decisions based on data.

Master handling of NULL values in SQL using specialized functions.

Learn to work with JSON data in SQL using specialized functions for querying and manipulation.

Challenge: Categorize Transactions

Solution

Challenge: Categorize Transactions

Solution