This Python script utilizes the sqlite3 library to create a SQLite database, store sales data, and query the database to find the top-selling products in each region for the year 2023 based on total sales revenue.
sales with sample data.quantity sold and the price of the product for the year 2023.region_id, product_id, and total_sales_revenue.#!/usr/bin/env python
import sqlite3
# Connect to SQLite3 database (it will create the database if it doesn't exist)
conn = sqlite3.connect('sales_data.db')
cursor = conn.cursor()
# Create the sales table
cursor.execute('''
CREATE TABLE IF NOT EXISTS sales (
transaction_id INT,
product_id INT,
region_id INT,
sale_date DATE,
quantity INT,
price DECIMAL(10, 2)
);
''')
# Insert the sample data
cursor.executemany('''
INSERT INTO sales VALUES (?, ?, ?, ?, ?, ?);
''', [
(1, 101, 1, '2023-01-05', 100, 10.50),
(2, 102, 1, '2023-02-10', 150, 15.75),
(3, 103, 1, '2023-03-15', 200, 20.00),
(4, 101, 2, '2023-01-05', 120, 11.25),
(5, 102, 2, '2023-02-10', 180, 18.50),
(6, 103, 2, '2023-03-15', 220, 22.75),
(7, 101, 3, '2023-01-05', 80, 9.75),
(8, 102, 3, '2023-02-10', 100, 12.25),
(9, 103, 3, '2023-03-15', 150, 17.00)
])
# Commit the transaction
conn.commit()
# SQL query to find the top-selling product in each region for the year 2023
query = '''
WITH
sales_2023 AS (
SELECT
product_id,
region_id,
SUM(quantity * price) AS total_sales_revenue
FROM
sales
WHERE
strftime('%Y', sale_date) = '2023'
GROUP BY
product_id, region_id
),
ranked_sales AS (
SELECT
product_id,
region_id,
total_sales_revenue,
RANK() OVER (PARTITION BY region_id ORDER BY total_sales_revenue DESC) AS rank
FROM
sales_2023
)
SELECT
region_id,
product_id,
total_sales_revenue
FROM
ranked_sales
WHERE
rank = 1;
'''
# Execute the query
cursor.execute(query)
results = cursor.fetchall()
# Display the results
print("Top-selling products in each region for 2023:")
for row in results:
print(f"Region ID: {row[0]}, Product ID: {row[1]}, Total Sales Revenue: {row[2]}")
# Close the connection
conn.close()Top-selling products in each region for 2023:
Region ID: 1, Product ID: 103, Total Sales Revenue: 4000
Region ID: 2, Product ID: 103, Total Sales Revenue: 5005.0
Region ID: 3, Product ID: 103, Total Sales Revenue: 2550The script begins by creating a connection to a SQLite database. If the database does not exist, it creates one named sales_data.db. After connecting to the database, the sales table is created with the specified columns: transaction_id, product_id, region_id, sale_date, quantity, and price. The sample data is then inserted into the table.
Once the data is inserted, the script runs an SQL query to calculate the total sales revenue for each product in each region for the year 2023. It uses a window function RANK() to rank the products by their total sales revenue within each region. The script then selects the product with the highest sales revenue in each region and displays the result.
The "top N per group" problem is one of the most common analytical queries, and window functions solve it cleanly. The inner query filters to 2023 and aggregates revenue per region and product. The window function then ranks products within each region, PARTITION BY region_id restarting the ranking for every region and ORDER BY total_sales_revenue DESC putting the highest revenue first. The outer query keeps rows with rank 1. A window function cannot appear directly in a WHERE clause because windows are computed after filtering and grouping, which is why the ranking sits in a subquery or common table expression.
SQLite has supported window functions since version 3.25, which ships with every currently supported Python release, so the script runs without any extra database server.
The choice of ranking function decides what happens when two products have identical revenue. RANK gives both rank 1 and skips the next number; DENSE_RANK gives both rank 1 without gaps; ROW_NUMBER assigns distinct numbers, so exactly one row per region is returned but which one is arbitrary unless the ORDER BY includes a tie-breaker such as product id. For reports, either show ties explicitly or add a deterministic tie-breaker.
In a production schema, the sales table would usually reference separate product and region tables by key, and the query would join them after ranking to show names. Ranking on keys first and joining afterwards keeps the window computation on a narrow, already aggregated result, which is cheaper than ranking a wide joined table. The same pattern applies unchanged in PostgreSQL, MySQL 8, SQL Server, Snowflake, and Spark SQL, since window functions are part of the SQL standard; only date functions and the optional QUALIFY shortcut differ between engines.
WITH revenue AS (
SELECT region_id, product_id, SUM(sales_amount) AS total_sales_revenue
FROM sales
WHERE sale_date >= '2023-01-01' AND sale_date < '2024-01-01'
GROUP BY region_id, product_id
),
ranked AS (
SELECT *, DENSE_RANK() OVER (PARTITION BY region_id
ORDER BY total_sales_revenue DESC) AS rnk
FROM revenue
)
SELECT region_id, product_id, total_sales_revenue, rnk
FROM ranked
WHERE rnk <= 3
ORDER BY region_id, rnk;Filtering on a date range rather than strftime('%Y', sale_date) = '2023' lets the database use an index on sale_date. Column names here follow the pattern of the script above; adjust them to your table.
import sqlite3
con = sqlite3.connect("sales.db")
con.execute("CREATE INDEX IF NOT EXISTS ix_sales_date ON sales(sale_date)")
for row in con.execute(open("top3.sql").read()):
print(row)| Approach | Notes |
|---|---|
| Window function (this page) | Clear, handles ties, supports top N |
| Correlated subquery with MAX | Works on old engines; slower on large tables |
| Join to grouped MAX | Portable; returns ties; only top 1 |
Snowflake / Databricks QUALIFY | Filters on window results without a subquery |
pandas groupby().nlargest() | Fine for small data already in memory |
PARTITION BY ranks across all regions instead of within each one.