Summary  
This chapter covers the COUNT() aggregate function in SQL, showing how to count rows in a table or count only those matching specific criteria using both COUNT(column) and COUNT(*).

General domain of usage  
relational database querying

`COUNT()` is a function that allows you to count rows in a table or count how many rows match certain criteria. 

Let's see an example:

To count the total number of rows in the table, we can use the following syntax:

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()
        # Create table and insert test data
        cls.cursor.execute("""
            DROP TABLE IF EXISTS country;
            CREATE TABLE country (
                id INTEGER PRIMARY KEY,
                name VARCHAR(100),
                continent VARCHAR(50),
                region VARCHAR(100),
                surfacearea FLOAT,
                capital VARCHAR(100),
                population INTEGER
            );
        """)
        # Insert test data
        countries = [
            (1, 'China', 'Asia', 'Eastern Asia', 9596961, 'Beijing', 1400000000),
            (2, 'India', 'Asia', 'Southern Asia', 3287263, 'New Delhi', 1300000000),
            (3, 'Japan', 'Asia', 'Eastern Asia', 377930, 'Tokyo', 126000000),
            (4, 'Nepal', 'Asia', 'Southern Asia', 147181, 'Kathmandu', 30000000),
            (5, 'Bhutan', 'Asia', 'Southern Asia', 38394, 'Thimphu', 800000), # under 1M
            (6, 'Russia', 'Europe', 'Eastern Europe', 17098242, 'Moscow', 144000000), # not Asia
            (7, 'Egypt', 'Africa', 'Northern Africa', 1002450, 'Cairo', 100000000), # not Asia
            (8, 'Singapore', 'Asia', 'South-Eastern Asia', 719, 'Singapore', 5700000), # over 1M
            (9, 'Maldives', 'Asia', 'Southern Asia', 298, 'Malé', 400000), # under 1M
            (10, 'Thailand', 'Asia', 'South-Eastern Asia', 513120, 'Bangkok', 69000000), # over 1M
            (11, 'Kazakhstan', 'Asia', 'Central Asia', 2724900, 'Nur-Sultan', 18500000), # over 1M
            (12, 'Georgia', 'Asia', 'Western Asia', 69700, 'Tbilisi', 4000000), # over 1M
            (13, 'FakeAsia', 'asia', None, 1000, 'FakeCity', 2000000), # lowercase asia, for test
        ]
        cls.cursor.executemany("""
            INSERT INTO country (id, name, continent, region, surfacearea, capital, population)
            VALUES (%s, %s, %s, %s, %s, %s, %s)
        """, countries)
        cls.connection.commit()

    @classmethod
    def tearDownClass(cls):
        cls.cursor.execute("DROP TABLE IF EXISTS country;")
        cls.connection.commit()
        cls.cursor.close()
        cls.connection.close()

    def test_count_asian_countries_population_over_1m(self):
        """Check if the query returns the correct count for Asian countries with population > 1,000,000."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        # The correct count: China, India, Japan, Nepal, Singapore, Thailand, Kazakhstan, Georgia (all Asia, >1M, not counting Bhutan/Maldives/FakeAsia)
        # But Nepal: 30M, Singapore: 5.7M, Thailand: 69M, Kazakhstan: 18.5M, Georgia: 4M, Japan: 126M, China: 1.4B, India: 1.3B
        # Exclude Bhutan (800k, <1M), Maldives (400k, <1M), FakeAsia (continent='asia', not 'Asia')
        expected_count = 8
        condition = (
            len(rows) == 1 and
            len(rows[0]) == 1 and
            rows[0][0] == expected_count
        )
        _dynamic_test(
            self,
            condition,
            "Correct count returned for Asian countries with population over 1,000,000.",
            f"Expected count {expected_count}, but got {rows[0][0] if rows and rows[0] else 'no result'}."
        )

    def test_returns_single_row(self):
        """Ensure only a single row is returned by the query."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        condition = len(rows) == 1
        _dynamic_test(
            self,
            condition,
            "Query returns exactly one row.",
            f"Query returned {len(rows)} rows instead of 1."
        )

    def test_returns_single_column(self):
        """Ensure only a single column is returned by the query."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        # Check at least one row and only one column
        condition = (len(rows) == 1 and len(rows[0]) == 1)
        _dynamic_test(
            self,
            condition,
            "Query returns exactly one column.",
            f"Query returned {len(rows[0]) if rows else 0} columns instead of 1."
        )

    def test_query_case_sensitive_for_asia(self):
        """Ensure the query does not match 'asia' in lowercase: only 'Asia' should be matched."""
        # The row with id=13 ('FakeAsia', 'asia', ...) should NOT be counted
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        expected_count = 8
        condition = (
            len(rows) == 1 and
            len(rows[0]) == 1 and
            rows[0][0] == expected_count
        )
        _dynamic_test(
            self,
            condition,
            "Query is case sensitive for 'Asia'.",
            f"Query counted rows with 'asia' (lowercase). Expected {expected_count}, got {rows[0][0] if rows and rows[0] else 'no result'}."
        )

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):
    import re
    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

Gain a solid understanding of SQL essentials tailored specifically for business intelligence. You will master querying, filtering, and aggregating business datasets through hands-on tasks designed to mirror real-world professional scenarios.

The COUNT() Function

The COUNT() Function

Рішення