Summary  
The chapter covers using SQL queries to perform data quality checks—such as detecting duplicate records and out-of-range values—by leveraging grouping and conditional filters.

General domain of usage  
Business analytics

Ensuring data quality is a foundational step in analytical reporting. When working with large datasets, especially in business analytics, you must check for issues such as **missing values**, **duplicate records**, and **out-of-range values**. These problems can distort your results, lead to faulty conclusions, and undermine trust in your insights. By writing SQL queries to identify and address these data quality issues, you can improve the reliability of your analyses.

You can use SQL to systematically check for these problems. For example, to find duplicate records in a customer table, you can group by key identifying columns and look for groups with more than one record.

Data quality checks are often driven by real business needs. Suppose the HR team is concerned that employees might be assigned to more than one department, which could lead to reporting inconsistencies or HR policy violations. You can write a query to identify employees who appear in multiple departments, ensuring the integrity of your HR data.

Detecting out-of-range values is another crucial check. For instance, in a sales table, negative or zero transaction amounts may indicate refunds, cancellations, or data entry errors. You should be able to quickly identify these records to investigate further or exclude them from certain analyses.

Common data quality issues in analytics include:
- **Missing values:** Data fields that are null or empty, which may affect computations or aggregations;
- **Duplicates:** Multiple records representing the same real-world entity, causing double-counting or skewed results;
- **Out-of-range values:** Data entries that fall outside expected or logical boundaries, indicating possible data entry errors or system problems.

Definition

-- Find potential duplicate customer records based on first name, last name, and email
SELECT
  first_name,
  last_name,
  email,
  COUNT(*) AS record_count
FROM customers
GROUP BY first_name, last_name, email
HAVING COUNT(*) < 2;

**Use case:** The HR team wants to ensure there are no employees assigned to multiple departments.

-- Identify employees with multiple department assignments
SELECT
  first_name,
  last_name,
  COUNT(DISTINCT department_id) AS department_count
FROM hr_employees
GROUP BY first_name, last_name
HAVING COUNT(DISTINCT department_id) > 1;

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_returns_all_columns(self):
        """Check that all columns from sales table are selected."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute("SELECT * FROM sales LIMIT 0;")
        expected_columns = [desc[0] for desc in self.cursor.description]
        self.cursor.execute(query)
        actual_columns = [desc[0] for desc in self.cursor.description]
        missing = set(expected_columns) - set(actual_columns)
        extra = set(actual_columns) - set(expected_columns)
        condition = (len(missing) == 0 and len(extra) == 0)
        failure_msg = "Missing columns: {} | Extra columns: {}".format(missing, extra)
        _dynamic_test(
            self,
            condition,
            "Query returns all columns from sales table.",
            failure_msg
        )

    def test_only_negative_or_zero_amount(self):
        """Check that only rows with total_amount <= 0 are returned."""
        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)
        if not df.empty:
            invalid = df[df['total_amount'] > 0]
            condition = invalid.empty
        else:
            condition = True
        failure_msg = "Returned rows with total_amount > 0."
        _dynamic_test(
            self,
            condition,
            "No rows with total_amount > 0 are returned.",
            failure_msg
        )

    def test_no_missing_negative_or_zero(self):
        """Check that all negative or zero total_amount sales are included."""
        with open("query.sql", "r") as file:
            query = file.read()
        # Get all expected rows
        self.cursor.execute("SELECT * FROM sales WHERE total_amount <= 0;")
        expected_rows = self.cursor.fetchall()
        expected_columns = [desc[0] for desc in self.cursor.description]
        expected_df = pd.DataFrame(expected_rows, columns=expected_columns)
        # Get user result
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)
        # Find missing rows
        missing = []
        for _, row in expected_df.iterrows():
            mask = (result_df[result_df.columns] == row.values).all(axis=1)
            if not mask.any():
                missing.append(row.to_dict())
        condition = len(missing) == 0
        if not condition:
            failure_msg = "Missing the following sales with total_amount <= 0:\\n" + "\\n".join(str(r) for r in missing)
        else:
            failure_msg = ""
        _dynamic_test(
            self,
            condition,
            "All sales with total_amount <= 0 are included.",
            failure_msg
        )

    def test_excludes_positive_amount(self):
        """Check that no rows with total_amount > 0 are present."""
        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)
        if not df.empty:
            condition = (df['total_amount'] <= 0).all()
        else:
            condition = True
        failure_msg = "Rows with total_amount > 0 are present in the result."
        _dynamic_test(
            self,
            condition,
            "No rows with total_amount > 0 are present in the result.",
            failure_msg
        )

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 advanced SQL techniques for building analytical reports. Learn to solve real-world business problems using multi-level aggregations, window functions, cohort and funnel analysis, KPI calculations, and more. Each chapter features theory, a realistic business use case, and a hands-on coding task.

Explore advanced SQL concepts and apply them to real-world analytical reporting scenarios across multiple business domains.

Data Quality Checks as Analytical Queries

Solução