21 - Advanced PostgreSQL Querying

from typing import List, Dict
import psycopg2
from psycopg2.extras import RealDictCursor

# Type-annotated connection function
def get_db_connection(config: Dict[str, str]) -> psycopg2.extensions.connection:
    """Connect to PostgreSQL database."""
    return psycopg2.connect(
        dbname=config["dbname"],
        user=config["user"],
        password=config["password"],
        host=config["host"],
        cursor_factory=RealDictCursor
    )

# Query using CTE
def query_sales_by_date(config: Dict[str, str]) -> List[Dict]:
    """Query total sales by date using CTE."""
    query = """
    WITH valid_transactions AS (
        SELECT product, price, quantity, date
        FROM transactions
        WHERE product IS NOT NULL
        AND price > 0
        AND quantity > 0
    )
    SELECT date, SUM(price * quantity) as total_sales
    FROM valid_transactions
    GROUP BY date
    ORDER BY date;
    """
    print(f"Executing query:\n{query}")  # Debug: print query for inspection
    conn = get_db_connection(config)
    cursor = conn.cursor()
    cursor.execute(query)
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    print(f"Query results: {results}")  # Debug: print results
    return results

# Example usage
config = {
    "dbname": "hijra_db",
    "user": "postgres",
    "password": "password",
    "host": "localhost"
}
results = query_sales_by_date(config)

from typing import List, Dict
import psycopg2
from psycopg2.extras import RealDictCursor

def query_running_total(config: Dict[str, str]) -> List[Dict]:
    """Query running total of sales by date."""
    query = """
    SELECT
        date,
        product,
        price * quantity as sale_amount,
        SUM(price * quantity) OVER (
            ORDER BY date
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) as running_total
    FROM transactions
    WHERE product IS NOT NULL
        AND price > 0
        AND quantity > 0
    ORDER BY date;
    """
    print(f"Executing query:\n{query}")  # Debug: print query
    conn = get_db_connection(config)
    cursor = conn.cursor()
    cursor.execute(query)
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    print(f"Query results: {results}")  # Debug: print results
    return results

from typing import List, Dict
import psycopg2
from psycopg2.extras import RealDictCursor

def query_jsonb_data(config: Dict[str, str]) -> List[Dict]:
    """Query transactions with JSONB metadata."""
    query = """
    SELECT
        product,
        price,
        quantity,
        metadata->>'customer_id' as customer_id
    FROM transactions
    WHERE metadata @> '{"customer_id": "C001"}'::jsonb
    AND price > 0;
    """
    print(f"Executing query:\n{query}")  # Debug: print query
    conn = get_db_connection(config)
    cursor = conn.cursor()
    cursor.execute(query)
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    print(f"Query results: {results}")  # Debug: print results
    return results

min_price: 10.0
max_quantity: 100
required_fields:
  - product
  - price
  - quantity
product_prefix: 'Halal'
max_decimals: 2
dbname: hijra_db
user: postgres
password: password
host: localhost

plt.savefig(plot_path, dpi=100)
plt.show()  # Temporary for debugging
plt.close()

# File: de-onboarding/utils.py (reused from Chapter 3)
from typing import Union, Dict

def is_numeric(s: str, max_decimals: int = 2) -> bool:
    """Check if string is a decimal number with up to max_decimals."""
    parts = s.split(".")
    if len(parts) != 2 or not parts[0].replace("-", "").isdigit() or not parts[1].isdigit():
        return False
    return len(parts[1]) <= max_decimals

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

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

def has_valid_decimals(x: Union[int, float, str], max_decimals: int) -> bool:
    """Check if value has valid decimal places."""
    return is_numeric(str(x), max_decimals)

def apply_valid_decimals(x: Union[int, float, str], max_decimals: int) -> bool:
    """Apply has_valid_decimals to a value."""
    return has_valid_decimals(x, max_decimals)

def is_integer(x: Union[int, float, str]) -> bool:
    """Check if value is an integer when converted to string."""
    return str(x).isdigit()

# File: de-onboarding/query_tool.py
from typing import List, Dict, Any
import pandas as pd
import yaml
import psycopg2
from psycopg2.extras import RealDictCursor
import matplotlib.pyplot as plt
import json
import os

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

def setup_database(config: Dict[str, Any], csv_path: str) -> None:
    """Set up transactions table and load CSV data."""
    conn = psycopg2.connect(
        dbname=config["dbname"],
        user=config["user"],
        password=config["password"],
        host=config["host"]
    )
    cursor = conn.cursor()

    # Create table with JSONB column
    cursor.execute("""
    DROP TABLE IF EXISTS transactions;
    CREATE TABLE transactions (
        transaction_id TEXT PRIMARY KEY,
        product TEXT,
        price REAL,
        quantity INTEGER,
        date DATE,
        metadata JSONB
    );
    """)

    # Load CSV and add sample JSONB data
    df = pd.read_csv(csv_path)
    for _, row in df.iterrows():
        metadata = {
            "customer_id": f"C{row['transaction_id'][-3:]}",
            "is_high_value": row["transaction_id"] == "T001"  # For Exercise 3
        }
        cursor.execute("""
        INSERT INTO transactions (transaction_id, product, price, quantity, date, metadata)
        VALUES (%s, %s, %s, %s, %s, %s);
        """, (
            row["transaction_id"],
            row["product"],
            row["price"],
            row["quantity"],
            row["date"],
            json.dumps(metadata)
        ))

    conn.commit()
    cursor.close()
    conn.close()
    print(f"Loaded {len(df)} rows into transactions table")

def query_daily_sales(config: Dict[str, Any]) -> List[Dict]:
    """Query daily sales by product using CTE."""
    query = """
    WITH valid_transactions AS (
        SELECT product, price, quantity, date
        FROM transactions
        WHERE product IS NOT NULL
        AND price > 0
        AND quantity > 0
        AND product LIKE %s
    )
    SELECT date, product, SUM(price * quantity) as total_sales
    FROM valid_transactions
    GROUP BY date, product
    ORDER BY date, product;
    """
    conn = psycopg2.connect(
        dbname=config["dbname"],
        user=config["user"],
        password=config["password"],
        host=config["host"],
        cursor_factory=RealDictCursor
    )
    cursor = conn.cursor()
    cursor.execute(query, (f"{config['product_prefix']}%",))
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    print(f"Daily sales results: {results}")  # Debug: print results
    return results

def query_running_total(config: Dict[str, Any]) -> List[Dict]:
    """Query running total of sales by date."""
    query = """
    SELECT
        date,
        product,
        price * quantity as sale_amount,
        SUM(price * quantity) OVER (
            ORDER BY date
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) as running_total
    FROM transactions
    WHERE product IS NOT NULL
        AND price > 0
        AND quantity > 0
        AND product LIKE %s
    ORDER BY date;
    """
    conn = psycopg2.connect(
        dbname=config["dbname"],
        user=config["user"],
        password=config["password"],
        host=config["host"],
        cursor_factory=RealDictCursor
    )
    cursor = conn.cursor()
    cursor.execute(query, (f"{config['product_prefix']}%",))
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    print(f"Running total results: {results}")  # Debug: print results
    return results

def query_customer_sales(config: Dict[str, Any]) -> List[Dict]:
    """Query sales for a specific customer using JSONB."""
    query = """
    SELECT
        product,
        price,
        quantity,
        metadata->>'customer_id' as customer_id
    FROM transactions
    WHERE metadata @> %s
    AND price > 0
    AND product IS NOT NULL;
    """
    conn = psycopg2.connect(
        dbname=config["dbname"],
        user=config["user"],
        password=config["password"],
        host=config["host"],
        cursor_factory=RealDictCursor
    )
    cursor = conn.cursor()
    cursor.execute(query, (json.dumps({"customer_id": "C001"}),))
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    print(f"Customer sales results: {results}")  # Debug: print results
    return results

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

def plot_daily_sales(daily_sales: List[Dict], plot_path: str) -> None:
    """Generate daily sales plot."""
    if not daily_sales:
        print("No data to plot")
        return

    df = pd.DataFrame(daily_sales)
    df["date"] = pd.to_datetime(df["date"])
    df = df.groupby("date")["total_sales"].sum().reset_index()

    plt.figure(figsize=(10, 6))
    plt.plot(df["date"], df["total_sales"], marker="o")
    plt.title("Daily Sales Trend")
    plt.xlabel("Date")
    plt.ylabel("Total Sales ($)")
    plt.grid(True)
    plt.tight_layout()
    plt.savefig(plot_path, dpi=100)
    plt.close()
    print(f"Plot saved to {plot_path}")
    print(f"File exists: {os.path.exists(plot_path)}")

def main() -> None:
    """Main function to run query tool."""
    csv_path = "data/transactions.csv"
    config_path = "data/config.yaml"
    json_path = "data/transaction_results.json"
    plot_path = "data/daily_sales.png"

    config = read_config(config_path)
    setup_database(config, csv_path)

    daily_sales = query_daily_sales(config)
    running_total = query_running_total(config)
    customer_sales = query_customer_sales(config)

    results = {
        "daily_sales": daily_sales,
        "running_total": running_total,
        "customer_sales": customer_sales
    }

    export_results(results, json_path)
    plot_daily_sales(daily_sales, plot_path)

    print("\nTransaction Report:")
    print(f"Daily Sales Records: {len(daily_sales)}")
    print(f"Running Total Records: {len(running_total)}")
    print(f"Customer Sales Records: {len(customer_sales)}")

if __name__ == "__main__":
    main()

# File: de-onboarding/test_query_tool.py
import pytest
from query_tool import read_config, setup_database, query_daily_sales, query_running_total, query_customer_sales
import psycopg2
from psycopg2.extras import RealDictCursor

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

@pytest.fixture
def setup_db(config):
    """Fixture to set up database."""
    setup_database(config, "data/transactions.csv")
    return config

@pytest.fixture
def empty_db(config):
    """Fixture to set up empty database."""
    conn = psycopg2.connect(
        dbname=config["dbname"],
        user=config["user"],
        password=config["password"],
        host=config["host"]
    )
    cursor = conn.cursor()
    cursor.execute("""
    DROP TABLE IF EXISTS transactions;
    CREATE TABLE transactions (
        transaction_id TEXT PRIMARY KEY,
        product TEXT,
        price REAL,
        quantity INTEGER,
        date DATE,
        metadata JSONB
    );
    """)
    conn.commit()
    cursor.close()
    conn.close()
    return config

def test_daily_sales(setup_db):
    """Test daily sales query."""
    results = query_daily_sales(setup_db)
    assert len(results) == 3  # 3 valid Halal products
    assert all("date" in r and "total_sales" in r for r in results)
    assert any(r["total_sales"] == 1999.98 for r in results)  # Halal Laptop

def test_running_total(setup_db):
    """Test running total query."""
    results = query_running_total(setup_db)
    assert len(results) == 3  # 3 valid Halal products
    assert all("running_total" in r for r in results)
    assert results[-1]["running_total"] == 2499.83  # Total of valid sales

def test_customer_sales(setup_db):
    """Test customer sales query."""
    results = query_customer_sales(setup_db)
    assert len(results) == 1
    assert results[0]["customer_id"] == "C001"

def test_empty_table(empty_db):
    """Test daily sales with empty table."""
    results = query_daily_sales(empty_db)
    print("Empty table returned:", results)  # Debug: print results
    assert results == []  # Explicitly check empty list

def test_invalid_jsonb(setup_db):
    """Test query with invalid JSONB data."""
    conn = psycopg2.connect(
        dbname=setup_db["dbname"],
        user=setup_db["user"],
        password=setup_db["password"],
        host=setup_db["host"]
    )
    cursor = conn.cursor()
    cursor.execute("""
    INSERT INTO transactions (transaction_id, product, price, quantity, date, metadata)
    VALUES (%s, %s, %s, %s, %s, %s);
    """, ("T006", "Halal Monitor", 299.99, 1, "2023-10-06", "{invalid}"))
    conn.commit()
    cursor.close()
    conn.close()

    results = query_customer_sales(setup_db)
    print("Invalid JSONB filtered:", results)  # Debug: print results
    assert len(results) == 1  # Only valid JSONB records returned
    assert results[0]["customer_id"] == "C001"  # No exceptions raised

{
  "daily_sales": [
    { "date": "2023-10-01", "product": "Halal Laptop", "total_sales": 1999.98 },
    { "date": "2023-10-02", "product": "Halal Mouse", "total_sales": 249.9 },
    { "date": "2023-10-03", "product": "Halal Keyboard", "total_sales": 249.95 }
  ],
  "running_total": [
    {
      "date": "2023-10-01",
      "product": "Halal Laptop",
      "sale_amount": 1999.98,
      "running_total": 1999.98
    },
    {
      "date": "2023-10-02",
      "product": "Halal Mouse",
      "sale_amount": 249.9,
      "running_total": 2249.88
    },
    {
      "date": "2023-10-03",
      "product": "Halal Keyboard",
      "sale_amount": 249.95,
      "running_total": 2499.83
    }
  ],
  "customer_sales": [
    {
      "product": "Halal Laptop",
      "price": 999.99,
      "quantity": 2,
      "customer_id": "C001"
    }
  ]
}

from typing import List, Dict
import psycopg2
from psycopg2.extras import RealDictCursor

def query_product_sales(config: Dict[str, Any]) -> List[Dict]:
    """Query total sales by product using CTE."""
    query = """
    WITH valid_transactions AS (
        SELECT product, price, quantity
        FROM transactions
        WHERE product LIKE %s
        AND price > 0
        AND quantity > 0
    )
    SELECT product, SUM(price * quantity) as total_sales
    FROM valid_transactions
    GROUP BY product
    ORDER BY total_sales DESC;
    """
    conn = psycopg2.connect(
        dbname=config["dbname"],
        user=config["user"],
        password=config["password"],
        host=config["host"],
        cursor_factory=RealDictCursor
    )
    cursor = conn.cursor()
    cursor.execute(query, (f"{config['product_prefix']}%",))
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    return results

from typing import List, Dict
import psycopg2
from psycopg2.extras import RealDictCursor

def query_product_rankings(config: Dict[str, Any]) -> List[Dict]:
    """Query product sales rankings."""
    query = """
    SELECT
        product,
        price * quantity as sale_amount,
        RANK() OVER (
            ORDER BY price * quantity DESC
        ) as sales_rank
    FROM transactions
    WHERE product LIKE %s
    AND price > 0
    AND quantity > 0
    ORDER BY sales_rank;
    """
    conn = psycopg2.connect(
        dbname=config["dbname"],
        user=config["user"],
        password=config["password"],
        host=config["host"],
        cursor_factory=RealDictCursor
    )
    cursor = conn.cursor()
    cursor.execute(query, (f"{config['product_prefix']}%",))
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    return results

from typing import List, Dict
import psycopg2
from psycopg2.extras import RealDictCursor
import json

def setup_ex3_database(config: Dict[str, Any]) -> None:
    """Set up transactions table for Exercise 3 with high-value customer data."""
    conn = psycopg2.connect(
        dbname=config["dbname"],
        user=config["user"],
        password=config["password"],
        host=config["host"]
    )
    cursor = conn.cursor()

    # Create table with JSONB column
    cursor.execute("""
    DROP TABLE IF EXISTS transactions;
    CREATE TABLE transactions (
        transaction_id TEXT PRIMARY KEY,
        product TEXT,
        price REAL,
        quantity INTEGER,
        date DATE,
        metadata JSONB
    );
    """)

    # Insert sample high-value customer data
    metadata = {"customer_id": "C001", "is_high_value": True}
    cursor.execute("""
    INSERT INTO transactions (transaction_id, product, price, quantity, date, metadata)
    VALUES (%s, %s, %s, %s, %s, %s);
    """, ("T001", "Halal Laptop", 999.99, 2, "2023-10-01", json.dumps(metadata)))

    conn.commit()
    cursor.close()
    conn.close()
    print("Loaded high-value customer data for Exercise 3")

def query_high_value_customers(config: Dict[str, Any]) -> List[Dict]:
    """Query transactions for high-value customers using JSONB."""
    setup_ex3_database(config)  # Seed data for exercise
    query = """
    SELECT
        product,
        metadata->>'customer_id' as customer_id,
        metadata->>'is_high_value' as is_high_value
    FROM transactions
    WHERE metadata @> '{"is_high_value": true}'::jsonb
    AND price > 0
    AND product IS NOT NULL;
    """
    conn = psycopg2.connect(
        dbname=config["dbname"],
        user=config["user"],
        password=config["password"],
        host=config["host"],
        cursor_factory=RealDictCursor
    )
    cursor = conn.cursor()
    cursor.execute(query)
    results = cursor.fetchall()
    cursor.close()
    conn.close()

    # Write conceptual explanation
    explanation = """
    JSONB is ideal for flexible, semi-structured data like customer metadata in Hijra Group’s analytics, enabling dynamic queries. Normalized tables are better for structured, performance-critical data like product prices, offering faster queries with indexes. Use JSONB for infrequent, flexible queries; use normalized tables for frequent, structured access.
    """
    with open("de-onboarding/ex3_concepts.txt", "w") as file:
        file.write(explanation)

    print(f"Explanation saved to de-onboarding/ex3_concepts.txt")
    return results

from typing import List, Dict
import matplotlib.pyplot as plt
import pandas as pd

def plot_running_total(running_total: List[Dict], plot_path: str) -> None:
    """Plot running total of sales."""
    if not running_total:
        print("No data to plot")
        return
    df = pd.DataFrame(running_total)
    df["date"] = pd.to_datetime(df["date"])
    plt.figure(figsize=(10, 6))
    plt.plot(df["date"], df["running_total"], marker="o")
    plt.title("Running Total Sales")
    plt.xlabel("Date")
    plt.ylabel("Running Total ($)")
    plt.grid(True)
    plt.tight_layout()
    plt.savefig(plot_path, dpi=100)
    plt.close()
    print(f"Plot saved to {plot_path}")

def query_buggy_sales(config: Dict[str, Any]) -> List[Dict]:
    query = """
    WITH valid_transactions AS (
        SELECT product, price, quantity, date
        FROM transactions
        WHERE product IS NOT NULL
    )
    SELECT date, SUM(price * quantity) as total_sales
    FROM valid_transactions
    GROUP BY price;  # Bug: Wrong column
    """
    conn = psycopg2.connect(
        dbname=config["dbname"],
        user=config["user"],
        password=config["password"],
        host=config["host"],
        cursor_factory=RealDictCursor
    )
    cursor = conn.cursor()
    cursor.execute(query)
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    return results

from typing import List, Dict
import psycopg2
from psycopg2.extras import RealDictCursor

def query_fixed_sales(config: Dict[str, Any]) -> List[Dict]:
    """Query total sales by date using CTE."""
    query = """
    WITH valid_transactions AS (
        SELECT product, price, quantity, date
        FROM transactions
        WHERE product IS NOT NULL
        AND price > 0
        AND quantity > 0
    )
    SELECT date, SUM(price * quantity) as total_sales
    FROM valid_transactions
    GROUP BY date
    ORDER BY date;
    """
    conn = psycopg2.connect(
        dbname=config["dbname"],
        user=config["user"],
        password=config["password"],
        host=config["host"],
        cursor_factory=RealDictCursor
    )
    cursor = conn.cursor()
    cursor.execute(query)
    results = cursor.fetchall()
    cursor.close()
    conn.close()
    return results

def explain_cte_vs_subquery() -> None:
    """Explain why CTEs are preferred over subqueries for daily sales reports."""
    answer = """
    CTEs improve readability and modularity for complex daily sales reports, making maintenance easier for Hijra Group’s analytics team. Unlike subqueries, CTEs allow reusable temporary result sets, simplifying query structure. They also enable PostgreSQL’s query planner to optimize joins efficiently, enhancing performance for large datasets.
    """
    with open("de-onboarding/ex6_concepts.txt", "w") as file:
        file.write(answer)
    print("Answer saved to de-onboarding/ex6_concepts.txt")