24 - Checkpoint 3B: Database Fundamentals II Review

WITH SalesCTE AS (
    SELECT product, price * quantity AS amount
    FROM sales
)
SELECT product, amount,
       RANK() OVER (ORDER BY amount DESC) AS sales_rank
FROM SalesCTE;

-- Joining sales with a products table
SELECT s.product, s.price, p.category
FROM sales s
JOIN products p ON s.product = p.product_name
WHERE s.product LIKE 'Halal%';

-- Select products with above-average price
SELECT product, price
FROM sales
WHERE price > (SELECT AVG(price) FROM sales)
AND product LIKE 'Halal%';

CREATE TABLE products (
    product_id INTEGER PRIMARY KEY,
    product_name TEXT
);
CREATE TABLE sales (
    sale_id INTEGER PRIMARY KEY,
    product_id INTEGER,
    price REAL,
    quantity INTEGER,
    FOREIGN KEY (product_id) REFERENCES products(product_id)
);

CREATE INDEX idx_product ON sales(product);
SELECT * FROM sales WHERE product = 'Halal Laptop';

from pydantic import BaseModel
class Sale(BaseModel):
    product: str
    price: float
    quantity: int

CREATE TABLE sales (
    product TEXT,
    price REAL,
    quantity INTEGER
);
INSERT INTO sales VALUES
('Halal Laptop', 999.99, 2),
('Halal Mouse', 24.99, 10),
('Halal Keyboard', 49.99, 5);

min_price: 10.0
max_quantity: 100
required_fields:
  - product
  - price
  - quantity
product_prefix: 'Halal'
max_decimals: 2

# File: de-onboarding/utils.py
from pydantic import BaseModel, validator
from typing import Dict, List
import yaml

class Config(BaseModel):
    min_price: float
    max_quantity: int
    required_fields: List[str]
    product_prefix: str
    max_decimals: int

    @validator('required_fields')
    def check_required_fields(cls, v):
        if not all(isinstance(f, str) for f in v):
            raise ValueError("All required fields must be strings")
        return v

def read_config(config_path: str) -> Dict:
    """Read YAML configuration."""
    print(f"Opening config: {config_path}")  # Debug
    with open(config_path, "r") as file:
        config = yaml.safe_load(file)
    config_model = Config(**config)
    print(f"Loaded config: {config_model.dict()}")  # Debug
    return config_model.dict()

def clean_string(s: str) -> str:
    """Strip whitespace from string."""
    return s.strip()

def is_numeric_value(x: float) -> bool:
    """Check if value is numeric."""
    return isinstance(x, (int, float))

def is_integer(x: float) -> bool:
    """Check if value is integer."""
    return isinstance(x, int) or (isinstance(x, float) and x.is_integer())

def has_valid_decimals(x: float, max_decimals: int) -> bool:
    """Check if value has valid decimal places."""
    if not is_numeric_value(x):
        return False
    parts = str(x).split(".")
    return len(parts) == 1 or len(parts[1]) <= max_decimals

# File: de-onboarding/database_tool.py
import sqlite3
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import json
import os
from typing import Tuple, Dict, List
from pydantic import BaseModel
import utils

class Sale(BaseModel):
    product: str
    price: float
    quantity: int

def connect_db(db_path: str) -> sqlite3.Connection:
    """Connect to SQLite database."""
    print(f"Connecting to: {db_path}")  # Debug
    conn = sqlite3.connect(db_path)
    return conn

def load_transactions(csv_path: str, config: Dict) -> pd.DataFrame:
    """Load and validate transactions CSV."""
    print(f"Loading CSV: {csv_path}")  # Debug
    df = pd.read_csv(csv_path)
    print("Initial DataFrame:")  # Debug
    print(df.head())

    required_fields = config["required_fields"]
    missing_fields = [f for f in required_fields if f not in df.columns]
    if missing_fields:
        print(f"Missing columns: {missing_fields}")  # Log
        return pd.DataFrame()

    df = df.dropna(subset=["product"])
    df = df[df["product"].str.startswith(config["product_prefix"])]
    df = df[df["quantity"].apply(utils.is_integer)]
    df["quantity"] = df["quantity"].astype(int)
    df = df[df["quantity"] <= config["max_quantity"]]
    df = df[df["price"].apply(utils.is_numeric_value)]
    df = df[df["price"] > 0]
    df = df[df["price"] >= config["min_price"]]
    df = df[df["price"].apply(lambda x: utils.has_valid_decimals(x, config["max_decimals"]))]

    print("Validated DataFrame:")  # Debug
    print(df)
    return df

def query_sales(conn: sqlite3.Connection, config: Dict) -> pd.DataFrame:
    """Query sales data with advanced SQL."""
    query = """
    WITH SalesCTE AS (
        SELECT product, price * quantity AS amount
        FROM sales
        WHERE product LIKE ? || '%'
            AND price >= ?
            AND quantity <= ?
            AND price = ROUND(price, 2)  -- Ensure up to 2 decimal places
    )
    SELECT product, amount,
           RANK() OVER (ORDER BY amount DESC) AS sales_rank
    FROM SalesCTE;
    """
    # Example PostgreSQL query (optional, test locally if PostgreSQL is installed):
    # query = """
    # WITH SalesCTE AS (
    #     SELECT product, price * quantity AS amount
    #     FROM sales
    #     WHERE product LIKE %s || '%'
    #         AND price >= %s
    #         AND quantity <= %s
    #         AND price = ROUND(price, 2)
    # )
    # SELECT product, amount,
    #        ROW_NUMBER() OVER (ORDER BY amount DESC) AS sales_rank
    # FROM SalesCTE;
    # """
    params = (config["product_prefix"], config["min_price"], config["max_quantity"])
    df = pd.read_sql_query(query, conn, params=params)
    print("Queried DataFrame:")  # Debug
    print(df)
    return df

def process_data(sales_df: pd.DataFrame, transactions_df: pd.DataFrame) -> Dict:
    """Process data and compute metrics."""
    if sales_df.empty and transactions_df.empty:
        print("No valid data")  # Log
        return {"total_sales": 0.0, "top_products": {}, "unique_products": []}

    # Combine data
    combined_df = pd.concat([sales_df[["product", "amount"]], transactions_df[["product", "amount"]]])
    combined_df["amount"] = combined_df["amount"].astype(float)

    total_sales = np.sum(combined_df["amount"].values)
    unique_products = combined_df["product"].unique().tolist()
    sales_by_product = combined_df.groupby("product")["amount"].sum()
    top_products = sales_by_product.sort_values(ascending=False).head(3).to_dict()

    print(f"Total Sales: {total_sales}")  # Debug
    print(f"Top Products: {top_products}")  # Debug
    return {
        "total_sales": float(total_sales),
        "unique_products": unique_products,
        "top_products": top_products
    }

def export_results(results: Dict, json_path: str) -> None:
    """Export results to JSON."""
    print(f"Writing to: {json_path}")  # Debug
    with open(json_path, "w") as file:
        json.dump(results, file, indent=2)
    print(f"Exported to: {json_path}")  # Confirm

def plot_sales(df: pd.DataFrame, plot_path: str) -> None:
    """Generate sales plot."""
    if df.empty:
        print("No data to plot")  # Log
        return

    plt.figure(figsize=(8, 6))
    plt.bar(df["product"], df["amount"])
    plt.title("Sales by Product")
    plt.xlabel("Product")
    plt.ylabel("Sales Amount ($)")
    plt.xticks(rotation=45)
    plt.grid(True)
    plt.tight_layout()
    plt.savefig(plot_path, dpi=100)
    plt.close()
    print(f"Plot saved to: {plot_path}")  # Confirm

def main() -> None:
    """Main function."""
    db_path = "data/sales.db"
    csv_path = "data/transactions.csv"
    config_path = "data/config.yaml"
    json_path = "data/report.json"
    plot_path = "data/sales_report.png"

    config = utils.read_config(config_path)
    conn = connect_db(db_path)
    sales_df = query_sales(conn, config)
    transactions_df = load_transactions(csv_path, config)
    transactions_df["amount"] = transactions_df["price"] * transactions_df["quantity"]

    results = process_data(sales_df, transactions_df)
    export_results(results, json_path)
    plot_sales(sales_df, plot_path)

    print("\nSales Report:")
    print(f"Total Sales: ${round(results['total_sales'], 2)}")
    print(f"Unique Products: {results['unique_products']}")
    print(f"Top Products: {results['top_products']}")

    conn.close()

if __name__ == "__main__":
    main()

# File: de-onboarding/test_database_tool.py
import pytest
import pandas as pd
import sqlite3
import os
from database_tool import connect_db, query_sales, load_transactions, process_data, export_results, plot_sales
from utils import read_config

@pytest.fixture
def setup_db(tmp_path):
    db_path = tmp_path / "test.db"
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute("""
    CREATE TABLE sales (
        product TEXT,
        price REAL,
        quantity INTEGER
    )
    """)
    cursor.executemany(
        "INSERT INTO sales VALUES (?, ?, ?)",
        [("Halal Laptop", 999.99, 2), ("Halal Mouse", 24.99, 10)]
    )
    conn.commit()
    yield str(db_path)
    conn.close()

@pytest.fixture
def config():
    return read_config("data/config.yaml")

def test_connect_db(setup_db):
    conn = connect_db(setup_db)
    assert isinstance(conn, sqlite3.Connection)
    conn.close()

def test_query_sales(setup_db, config):
    conn = connect_db(setup_db)
    df = query_sales(conn, config)
    assert not df.empty
    assert "product" in df.columns
    assert "amount" in df.columns
    assert df["amount"].iloc[0] == 1999.98  # Halal Laptop
    conn.close()

def test_load_transactions(config):
    df = load_transactions("data/transactions.csv", config)
    assert not df.empty
    assert df["product"].str.startswith("Halal").all()
    assert df["quantity"].max() <= config["max_quantity"]

def test_process_data():
    sales_df = pd.DataFrame({
        "product": ["Halal Laptop"],
        "amount": [1999.98]
    })
    transactions_df = pd.DataFrame({
        "product": ["Halal Mouse"],
        "amount": [249.90]
    })
    results = process_data(sales_df, transactions_df)
    assert results["total_sales"] == 2249.88
    assert "Halal Laptop" in results["unique_products"]
    assert results["top_products"]["Halal Laptop"] == 1999.98

def test_export_results(tmp_path):
    results = {"total_sales": 2499.83, "unique_products": ["Halal Laptop"], "top_products": {"Halal Laptop": 1999.98}}
    json_path = tmp_path / "test.json"
    export_results(results, str(json_path))
    assert os.path.exists(json_path)

def test_plot_sales(tmp_path):
    df = pd.DataFrame({
        "product": ["Halal Laptop"],
        "amount": [1999.98]
    })
    plot_path = tmp_path / "test.png"
    plot_sales(df, str(plot_path))
    assert os.path.exists(plot_path)

{
  "total_sales": 4999.66,
  "unique_products": ["Halal Laptop", "Halal Mouse", "Halal Keyboard"],
  "top_products": {
    "Halal Laptop": 3999.96,
    "Halal Mouse": 499.8,
    "Halal Keyboard": 499.9
  }
}

def query_sales(conn):
    query = "SELECT product, price * quantity AS amount FROM sales WHERE product LIKE ?"
    params = ("Halal%", "invalid_param")  # Bug: Too many parameters
    return pd.read_sql_query(query, conn, params=params)

import sqlite3
import pandas as pd

def query_sales(db_path: str) -> pd.DataFrame:
    conn = sqlite3.connect(db_path)
    query = """
    WITH SalesCTE AS (
        SELECT product, price * quantity AS amount
        FROM sales
        WHERE product LIKE 'Halal%'
    )
    SELECT product, amount,
           RANK() OVER (ORDER BY amount DESC) AS sales_rank
    FROM SalesCTE;
    """
    df = pd.read_sql_query(query, conn)
    conn.close()
    return df

print(query_sales("data/sales.db"))

import pandas as pd
from pydantic import BaseModel
from typing import Dict
import utils

class Transaction(BaseModel):
    product: str
    price: float
    quantity: int

def validate_transactions(csv_path: str, config: Dict) -> pd.DataFrame:
    df = pd.read_csv(csv_path)
    # Filter for string products starting with prefix
    df = df[df["product"].apply(lambda x: isinstance(x, str) and x.startswith(config["product_prefix"]))]
    df = df[df["quantity"].apply(utils.is_integer)]
    df["quantity"] = df["quantity"].astype(int)
    df = df[df["price"].apply(utils.is_numeric_value)]
    for _, row in df.iterrows():
        Transaction(**row[["product", "price", "quantity"]].to_dict())
    return df

config = utils.read_config("data/config.yaml")
print(validate_transactions("data/transactions_invalid.csv", config))

import sqlite3
import time
import random

def optimize_index(db_path: str, output_path: str) -> Tuple[float, float]:
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()

    # Insert 1000 rows
    products = ["Halal Laptop", "Halal Mouse", "Halal Keyboard"]
    rows = [(random.choice(products), round(random.uniform(10.0, 1000.0), 2), random.randint(1, 100)) for _ in range(1000)]
    cursor.executemany("INSERT INTO sales (product, price, quantity) VALUES (?, ?, ?)", rows)
    conn.commit()

    query = "SELECT * FROM sales WHERE product = 'Halal Laptop'"
    start = time.time()
    cursor.execute(query)
    cursor.fetchall()
    before = time.time() - start

    cursor.execute("CREATE INDEX IF NOT EXISTS idx_product ON sales(product)")
    start = time.time()
    cursor.execute(query)
    cursor.fetchall()
    after = time.time() - start

    with open(output_path, "w") as f:
        f.write(f"Query Time Before Index: {before:.3f}s\nQuery Time After Index: {after:.3f}s\n")

    conn.close()
    return before, after

before, after = optimize_index("data/sales.db", "data/ex3_benchmark.txt")
print(f"Results saved to data/ex3_benchmark.txt")

import sqlite3
import pandas as pd
import matplotlib.pyplot as plt

def plot_sales(db_path: str, plot_path: str) -> None:
    conn = sqlite3.connect(db_path)
    df = pd.read_sql_query("SELECT product, price * quantity AS amount FROM sales WHERE product LIKE 'Halal%'", conn)
    plt.figure(figsize=(8, 6))
    plt.bar(df["product"], df["amount"])
    plt.title("Sales by Product")
    plt.xlabel("Product")
    plt.ylabel("Sales Amount ($)")
    plt.xticks(rotation=45)
    plt.grid(True)
    plt.tight_layout()
    plt.savefig(plot_path, dpi=100)
    plt.close()
    print(f"Plot saved to: {plot_path}")

plot_sales("data/sales.db", "data/ex4_plot.png")

import sqlite3
import pandas as pd

def query_sales(db_path: str) -> pd.DataFrame:
    conn = sqlite3.connect(db_path)
    query = "SELECT product, price * quantity AS amount FROM sales WHERE product LIKE ?"
    params = ("Halal%",)  # Fix: Correct number of parameters
    df = pd.read_sql_query(query, conn, params=params)
    conn.close()
    return df

print(query_sales("data/sales.db"))

# Save this explanation to data/ex6_concepts.txt
explanation = """
B-Tree indexes reduce query time from O(n) to O(log n) because they organize data in a balanced tree structure, allowing logarithmic-time lookups. For example, querying `sales` for `product = 'Halal Laptop'` without an index scans all rows (O(n)). With a B-Tree index on `product`, the database traverses the tree, reducing comparisons to O(log n).
"""
with open("data/ex6_concepts.txt", "w") as file:
    file.write(explanation)
print("Explanation saved to data/ex6_concepts.txt")