To solve this task, you need to identify the highest sale amount and the corresponding employee for each region in the `sales` table. You will use the `ROW_NUMBER()` window function to achieve this efficiently.

### Step-by-step approach

1. **Assign a row number within each region:**
   - Use `ROW_NUMBER()` as a window function to assign a unique number to each sale within its region.
   - Partition the data by the `region` column so that numbering restarts for each region.
   - Order the rows within each region by the `amount` column in descending order; this places the highest sale at the top for each region.

2. **Select only the top sale per region:**
   - After assigning row numbers, the row with `ROW_NUMBER() = 1` in each region will represent the highest sale for that region.
   - Use a subquery or common table expression (CTE) to filter and keep only those rows where the row number is 1.

3. **Return the required columns:**
   - Select the `region`, `employee_id`, and `amount` columns for the top sale in each region.



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_one_row_per_region(self):
        """Check that the result contains one row per unique region."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT DISTINCT region FROM sales")
        expected_regions = set(row[0] for row in self.cursor.fetchall())
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        result_regions = set(row[0] for row in rows)
        _dynamic_test(
            self,
            result_regions == expected_regions,
            "Result contains one row per unique region.",
            f"Result regions {result_regions} do not match expected regions {expected_regions}."
        )

    def test_highest_sale_amount_per_region(self):
        """Check that each row shows the highest sale amount for its region."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        self.cursor.execute("SELECT region, MAX(amount) FROM sales GROUP BY region")
        max_amounts = dict(self.cursor.fetchall())
        failed = []
        for row in rows:
            region, employee_id, amount = row
            if float(amount) != float(max_amounts[region]):
                failed.append((region, amount, max_amounts[region]))
        condition = len(failed) == 0
        if not condition:
            msg = "\\n".join([f"{r}: got {a}, expected {e}" for r, a, e in failed])
            failure_msg = f"Not all rows show the highest sale amount per region:\\n{msg}"
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "Each row shows the highest sale amount for its region.",
            failure_msg
        )

    def test_only_one_row_per_region_even_on_tie(self):
        """Check that only one row per region is returned, even if there are ties."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        region_counts = {}
        for row in rows:
            region = row[0]
            region_counts[region] = region_counts.get(region, 0) + 1
        failed = [r for r, c in region_counts.items() if c != 1]
        condition = len(failed) == 0
        if not condition:
            failure_msg = f"Regions with more than one row: {failed}"
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "Only one row per region is returned, even if there are ties.",
            failure_msg
        )

    def test_uses_row_number_window_function(self):
        """Check that the query uses ROW_NUMBER() OVER (PARTITION BY region ORDER BY amount DESC)."""
        with open("query.sql", "r") as file:
            query = file.read().lower().replace('\\n', ' ').replace('\\r', ' ')
        has_row_number = (
            "row_number" in query
            and "over" in query
            and "partition by" in query
            and "region" in query
            and "order by" in query
            and "amount" in query
            and "desc" in query
        )
        _dynamic_test(
            self,
            has_row_number,
            "Query uses ROW_NUMBER() OVER (PARTITION BY region ORDER BY amount DESC).",
            "Query does not use ROW_NUMBER() OVER (PARTITION BY region ORDER BY amount DESC)."
        )


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 art of analyzing data with SQL window functions. This course guides you through the core concepts, syntax, and practical applications of window functions, empowering you to perform advanced analytics and reporting directly in SQL.

Explore the foundational concepts of window functions, their syntax, and how they differ from traditional aggregate functions. Learn to apply basic window functions for analytics.

Dive into ranking, percentiles, and analytic functions. Learn to use ROW_NUMBER, RANK, DENSE_RANK, and NTILE for advanced reporting.

Apply window functions to solve complex business problems, including advanced framing, multi-function queries, and real-world analytics.

Challenge: Top Sales per Region

Step-by-step approach

Oplossing

Challenge: Top Sales per Region

Step-by-step approach

Oplossing