Summary  
This chapter explains how to use the SQL LIMIT clause to restrict the number of rows returned by a SELECT query, showing its syntax and placement at the end of the query.

General domain of usage  
Relational database querying

We know that the `SELECT` operator can retrieve all rows from a table of all or the specific columns. However, what if we only need to fetch a specific number of rows?

We can do it using the `LIMIT` clause. It is always used at the end of the query. Here is the syntax for it:

```sql
SELECT columns
FROM table 
LIMIT number_of_rows;
```

In the example below, we extract the first 7 rows (in our case, **the capitals of the countries**) from the column:

import unittest
import psycopg2
import pandas as pd
import re

class TestTask(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 and populate the country table
        cls.cursor.execute("DROP TABLE IF EXISTS country;")
        cls.cursor.execute("""
            CREATE TABLE country (
                id serial PRIMARY KEY,
                name varchar(100),
                continent varchar(50)
            );
        """)
        # Insert at least 3 distinct continents
        cls.cursor.executemany(
            "INSERT INTO country (name, continent) VALUES (%s, %s);",
            [
                ("France", "Europe"),
                ("Nigeria", "Africa"),
                ("Brazil", "South America"),
                ("China", "Asia"),
                ("Egypt", "Africa"),
                ("Germany", "Europe")
            ]
        )
        cls.connection.commit()

    def setUp(self):
        with open("query.sql", "r") as f:
            self.query = f.read()

    def test_returns_three_distinct_continents(self):
        """Check that query returns exactly 3 distinct continent values."""
        try:
            self.cursor.execute(self.query)
            rows = self.cursor.fetchall()
            continents = [row[0] for row in rows]
            unique_continents = set([c for c in continents if c is not None])
            correct_length = len(rows) == 3
            correct_distinct = len(unique_continents) == 3
            _dynamic_test(
                self,
                correct_length and correct_distinct,
                "Query returns exactly 3 distinct continent values.",
                f"Query must return exactly 3 distinct continent values. Got {len(rows)} rows and {len(unique_continents)} unique continents."
            )
        except Exception as e:
            self.connection.rollback()
            _dynamic_test(self, False, "", f"Query execution error: {e}")

    def test_select_distinct_used(self):
        """Check if SELECT DISTINCT is used on the continent column."""
        query = self.query.lower().replace("\t", " ")
        
        pattern = r"select\s+distinct(\s*\(|\s+)continent"
        has_select_distinct = re.search(pattern, query) is not None
        _dynamic_test(
            self,
            has_select_distinct,
            "Query uses SELECT DISTINCT on the continent column.",
            "Query must use SELECT DISTINCT on the continent column."
        )

    def test_output_column_name(self):
        """Check that the output column is named 'continent'."""
        try:
            self.cursor.execute(self.query)
            columns = [desc[0].lower() for desc in self.cursor.description]
            _dynamic_test(
                self,
                columns == ["continent"],
                "Output column is named 'continent'.",
                f"Output column should be named 'continent', got {columns}."
            )
        except Exception as e:
            self.connection.rollback()
            _dynamic_test(self, False, "", f"Query execution error: {e}")

if __name__ == "__main__":
    unittest.main()

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

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.

Limiting Results

Limiting Results

Løsning