17 - Python and PostgreSQL Integration

# Start PostgreSQL service (example for Ubuntu)
sudo service postgresql start

# Access psql as postgres user
psql -U postgres

# Create database
CREATE DATABASE onboarding;

# Connect to database
\c onboarding

# Create sales table
CREATE TABLE sales (
    id SERIAL PRIMARY KEY,
    product VARCHAR(255) NOT NULL,
    price DECIMAL(10, 2) NOT NULL CHECK (price > 0),
    quantity INTEGER NOT NULL CHECK (quantity > 0)
);

# Verify table creation
\dt
# Expected: Lists "sales" table

# Exit psql
\q

from typing import Optional  # For type hints
import psycopg2  # PostgreSQL adapter

def connect_db(dbname: str, user: str, password: str, host: str, port: str) -> Optional[psycopg2.extensions.connection]:
    """Connect to PostgreSQL database."""
    print(f"Connecting to database: {dbname}")  # Debug
    try:
        conn = psycopg2.connect(
            dbname=dbname,
            user=user,
            password=password,
            host=host,
            port=port
        )
        if conn.closed == 0:  # Verify connection is active
            print("Connected successfully")  # Debug
            return conn
        else:
            print("Connection is closed")  # Debug
            conn.close()
            return None
    except psycopg2.OperationalError as e:
        print(f"Connection failed: {e}")  # Log error
        return None
    # Note: Checking conn.closed ensures the connection is active, critical for reliable database operations.

from typing import List, Tuple  # For type hints
import psycopg2  # PostgreSQL adapter

def fetch_sales(conn: psycopg2.extensions.connection) -> List[Tuple[int, str, float, int]]:
    """Fetch all sales records."""
    cursor = conn.cursor()  # Create cursor
    cursor.execute("SELECT id, product, price, quantity FROM sales")  # Execute query
    results = cursor.fetchall()  # Fetch results
    cursor.close()  # Close cursor
    print(f"Fetched {len(results)} records")  # Debug
    return results  # Return list of tuples

# Example usage
if __name__ == "__main__":
    try:
        conn = connect_db("onboarding", "postgres", "", "localhost", "5432")
        if conn:
            sales = fetch_sales(conn)
            print("Sales:", sales)  # Debug
            conn.close()
    except psycopg2.Error as e:
        print(f"Query error: {e}")  # Debug
    # Note: This example simplifies error handling for clarity; production code (e.g., micro-project) includes robust try/except.

from pydantic import BaseModel, PositiveFloat, PositiveInt  # Pydantic types
from typing import Optional  # For optional fields

class Sale(BaseModel):
    """Pydantic model for sales data."""
    product: str
    price: PositiveFloat
    quantity: PositiveInt
    id: Optional[int] = None  # Optional for database ID

# Example usage
if __name__ == "__main__":
    sale = Sale(product="Halal Laptop", price=999.99, quantity=2)
    print("Valid sale:", sale)  # Debug
    # Invalid example
    try:
        Sale(product="Halal Mouse", price=-24.99, quantity=10)
    except ValueError as e:
        print("Validation error:", e)  # Debug

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

# File: de-onboarding/utils.py (updated from Chapter 3)
def is_numeric(s: str, max_decimals: int = 2) -> bool:  # Check if string is a decimal number
    """Check if string is a decimal number with up to max_decimals."""
    parts = s.split(".")  # Split on decimal point
    if len(parts) != 2 or not parts[0].isdigit() or not parts[1].isdigit():
        return False  # Invalid format (rejects negative numbers)
    return len(parts[1]) <= max_decimals  # Check decimal places

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

def is_numeric_value(x: object) -> bool:  # Check if value is numeric
    """Check if value is an integer or float."""
    return isinstance(x, (int, float))  # Return True for numeric types

def has_valid_decimals(x: float, max_decimals: int) -> bool:  # Check decimal places
    """Check if value has valid decimal places."""
    return is_numeric(str(x), max_decimals)  # Use is_numeric for validation

def apply_valid_decimals(x: float, max_decimals: int) -> bool:  # Apply decimal validation
    """Apply has_valid_decimals to a value."""
    return has_valid_decimals(x, max_decimals)

def is_integer(x: object) -> bool:  # Check if value is an integer
    """Check if value is an integer when converted to string."""
    return str(x).isdigit()  # Return True for integer strings

# File: de-onboarding/sales_pipeline.py
from typing import Dict, List, Optional, Tuple  # For type hints
import pandas as pd  # For DataFrame operations
import yaml  # For YAML parsing
import json  # For JSON export
import psycopg2  # PostgreSQL adapter
from pydantic import BaseModel, PositiveFloat, PositiveInt  # Pydantic validation
import utils  # Custom utilities

# Pydantic model for sales data
class Sale(BaseModel):
    """Pydantic model for sales data."""
    product: str
    price: PositiveFloat
    quantity: PositiveInt
    id: Optional[int] = None  # Optional for database ID

# Read YAML configuration
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)
    print(f"Loaded config: {config}")  # Debug
    return config

# Connect to PostgreSQL
def connect_db(dbname: str, user: str, password: str, host: str, port: str) -> Optional[psycopg2.extensions.connection]:
    """Connect to PostgreSQL database."""
    print(f"Connecting to database: {dbname}")  # Debug
    try:
        conn = psycopg2.connect(
            dbname=dbname,
            user=user,
            password=password,
            host=host,
            port=port
        )
        if conn.closed == 0:  # Verify connection is active
            print("Connected successfully")  # Debug
            return conn
        else:
            print("Connection is closed")  # Debug
            conn.close()
            return None
    except psycopg2.OperationalError as e:
        print(f"Connection failed: {e}")  # Log error
        return None

# Load and validate sales data
def load_and_validate_sales(csv_path: str, config: Dict) -> Tuple[List[Sale], int, int]:
    """Load sales CSV and validate with Pydantic."""
    print(f"Loading CSV: {csv_path}")  # Debug
    df = pd.read_csv(csv_path)  # Load CSV
    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 error
        return [], 0, len(df)

    valid_sales: List[Sale] = []
    for _, row in df.iterrows():
        try:
            # Clean and validate data
            product = str(row["product"]) if pd.notna(row["product"]) else ""
            price = float(row["price"]) if pd.notna(row["price"]) else 0.0
            quantity = int(row["quantity"]) if pd.notna(row["quantity"]) and utils.is_integer(row["quantity"]) else 0

            if not product.startswith(config["product_prefix"]):
                print(f"Invalid sale: product lacks '{config['product_prefix']}' prefix: {row.to_dict()}")  # Log
                continue
            if price < config["min_price"] or price <= 0:
                print(f"Invalid sale: price too low or negative: {row.to_dict()}")  # Log
                continue
            if quantity > config["max_quantity"]:
                print(f"Invalid sale: quantity too high: {row.to_dict()}")  # Log
                continue
            if not utils.apply_valid_decimals(price, config["max_decimals"]):
                print(f"Invalid sale: too many decimals: {row.to_dict()}")  # Log
                continue

            sale = Sale(product=product, price=price, quantity=quantity)
            valid_sales.append(sale)
        except (ValueError, TypeError) as e:
            print(f"Invalid sale: {row.to_dict()}, error: {e}")  # Log
            continue

    total_records = len(df)
    print(f"Validated {len(valid_sales)} sales")  # Debug
    return valid_sales, len(valid_sales), total_records

# Insert sales into PostgreSQL
def insert_sales(conn: psycopg2.extensions.connection, sales: List[Sale]) -> int:
    """Insert valid sales into PostgreSQL."""
    if not sales:
        print("No sales to insert")  # Log
        return 0

    cursor = conn.cursor()
    inserted = 0
    for sale in sales:
        try:
            cursor.execute(
                "INSERT INTO sales (product, price, quantity) VALUES (%s, %s, %s) RETURNING id",
                (sale.product, sale.price, sale.quantity)
            )
            sale.id = cursor.fetchone()[0]  # Update ID
            inserted += 1
        except psycopg2.Error as e:
            print(f"Insert failed: {sale}, error: {e}")  # Log
            conn.rollback()
            continue
    conn.commit()
    cursor.close()
    print(f"Inserted {inserted} sales")  # Debug
    return inserted

# Query total sales
def query_total_sales(conn: psycopg2.extensions.connection) -> float:
    """Query total sales amount."""
    cursor = conn.cursor()
    cursor.execute("SELECT SUM(price * quantity) FROM sales")
    total = cursor.fetchone()[0] or 0.0
    cursor.close()
    print(f"Total sales: {total}")  # Debug
    return float(total)

# Export results
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 results to {json_path}")  # Confirm

# Main function
def main() -> None:
    """Main function to process sales data."""
    csv_path = "data/sales.csv"
    config_path = "data/config.yaml"
    json_path = "data/sales_results.json"

    config = read_config(config_path)
    sales, valid_sales, total_records = load_and_validate_sales(csv_path, config)

    conn = connect_db("onboarding", "postgres", "", "localhost", "5432")
    if not conn:
        print("Exiting due to connection failure")
        return

    try:
        inserted = insert_sales(conn, sales)
        total_sales = query_total_sales(conn)
        results = {
            "total_sales": total_sales,
            "valid_sales": valid_sales,
            "invalid_sales": total_records - valid_sales,
            "inserted_sales": inserted
        }
        export_results(results, json_path)

        print("\nSales Report:")
        print(f"Total Records Processed: {total_records}")
        print(f"Valid Sales: {valid_sales}")
        print(f"Invalid Sales: {total_records - valid_sales}")
        print(f"Inserted Sales: {inserted}")
        print(f"Total Sales: ${round(total_sales, 2)}")
        print("Processing completed")
    finally:
        conn.close()
        print("Connection closed")

if __name__ == "__main__":
    main()

# File: de-onboarding/test_sales_pipeline.py
from typing import List
import pytest
import psycopg2
from sales_pipeline import connect_db, insert_sales, query_total_sales, load_and_validate_sales, Sale

@pytest.fixture
def db_conn():
    """Fixture for database connection."""
    conn = connect_db("onboarding", "postgres", "", "localhost", "5432")
    yield conn
    if conn:
        cursor = conn.cursor()
        cursor.execute("DELETE FROM sales")  # Clean up; commit ensures changes persist
        conn.commit()
        cursor.close()
        conn.close()

def test_insert_and_query_sales(db_conn):
    """Test inserting and querying sales."""
    sales = [
        Sale(product="Halal Laptop", price=999.99, quantity=2),
        Sale(product="Halal Mouse", price=24.99, quantity=10)
    ]
    inserted = insert_sales(db_conn, sales)
    assert inserted == 2, f"Expected 2 inserted, got {inserted}"

    total_sales = query_total_sales(db_conn)
    expected = 999.99 * 2 + 24.99 * 10
    assert abs(total_sales - expected) < 0.01, f"Expected {expected}, got {total_sales}"

def test_empty_sales(db_conn):
    """Test inserting empty sales list."""
    inserted = insert_sales(db_conn, [])
    assert inserted == 0, f"Expected 0 inserted, got {inserted}"
    total_sales = query_total_sales(db_conn)
    assert total_sales == 0.0, f"Expected 0.0, got {total_sales}"

def test_missing_table():
    """Test inserting with missing table."""
    conn = connect_db("onboarding", "postgres", "", "localhost", "5432")
    if not conn:
        pytest.skip("No database connection")
    cursor = conn.cursor()
    cursor.execute("DROP TABLE IF EXISTS sales")
    conn.commit()
    sales = [Sale(product="Halal Laptop", price=999.99, quantity=2)]
    with pytest.raises(psycopg2.ProgrammingError, match="relation \"sales\" does not exist"):
        insert_sales(conn, sales)
    conn.close()

def test_invalid_data():
    """Test validation of invalid sales data from file."""
    config = {
        "min_price": 10.0,
        "max_quantity": 100,
        "required_fields": ["product", "price", "quantity"],
        "product_prefix": "Halal",
        "max_decimals": 2
    }
    sales, valid_sales, total_records = load_and_validate_sales("data/invalid.csv", config)
    assert valid_sales == 1, f"Expected 1 valid sale, got {valid_sales}"
    assert total_records == 2, f"Expected 2 total records, got {total_records}"

{
  "total_sales": 2499.83,
  "valid_sales": 3,
  "invalid_sales": 3,
  "inserted_sales": 3
}

sales = [Sale(product="Halal Laptop", price=999.99, quantity=2)]

data = [
    {"product": "Halal Mouse", "price": 24.99, "quantity": 10},
    {"product": "Monitor", "price": 199.99, "quantity": 2}
]

import psycopg2
def connect_and_query():
    conn = psycopg2.connect(dbname="onboarding", user="postgres", password="", host="localhost", port="5432")
    cursor = conn.cursor()
    cursor.execute("SELECT COUNT(*) FROM sales")
    count = cursor.fetchone()[0]
    cursor.close()
    return count

sale = Sale(product="Halal Keyboard", price=49.99, quantity=5)

from typing import Optional
import psycopg2

def connect_db(dbname: str, user: str, password: str, host: str, port: str) -> Optional[psycopg2.extensions.connection]:
    """Connect to PostgreSQL database."""
    print(f"Connecting to database: {dbname}")
    try:
        conn = psycopg2.connect(dbname=dbname, user=user, password=password, host=host, port=port)
        if conn.closed == 0:
            print("Connection object created")
            return conn
        else:
            print("Connection is closed")
            conn.close()
            return None
    except psycopg2.OperationalError as e:
        print(f"Connection failed: {e}")
        return None

# Test
if __name__ == "__main__":
    try:
        conn = connect_db("onboarding", "postgres", "", "localhost", "5432")
        if conn:
            conn.close()
    except psycopg2.Error as e:
        print(f"Connection error: {e}")

from typing import List
import psycopg2
from pydantic import BaseModel, PositiveFloat, PositiveInt

class Sale(BaseModel):
    product: str
    price: PositiveFloat
    quantity: PositiveInt

def insert_sales(conn: psycopg2.extensions.connection, sales: List[Sale]) -> int:
    """Insert sales into PostgreSQL."""
    cursor = conn.cursor()
    inserted = 0
    for sale in sales:
        cursor.execute(
            "INSERT INTO sales (product, price, quantity) VALUES (%s, %s, %s)",
            (sale.product, sale.price, sale.quantity)
        )
        inserted += 1
    conn.commit()
    cursor.close()
    print(f"Inserted {inserted} sales")
    return inserted

# Test
if __name__ == "__main__":
    conn = connect_db("onboarding", "postgres", "", "localhost", "5432")
    sales = [Sale(product="Halal Laptop", price=999.99, quantity=2)]
    insert_sales(conn, sales)
    conn.close()

import psycopg2

def query_avg_price(conn: psycopg2.extensions.connection) -> float:
    """Query average price of sales."""
    cursor = conn.cursor()
    cursor.execute("SELECT AVG(price) FROM sales")
    avg_price = cursor.fetchone()[0] or 0.0
    cursor.close()
    print(f"Average price: {avg_price}")
    return float(avg_price)

# Test
if __name__ == "__main__":
    conn = connect_db("onboarding", "postgres", "", "localhost", "5432")
    avg_price = query_avg_price(conn)
    conn.close()

from typing import List, Dict
from pydantic import BaseModel, PositiveFloat, PositiveInt, ValidationError

class Sale(BaseModel):
    product: str
    price: PositiveFloat
    quantity: PositiveInt

def validate_sales(data: List[Dict], prefix: str) -> List[Sale]:
    """Validate sales data with Pydantic."""
    valid_sales = []
    for item in data:
        try:
            if not item["product"].startswith(prefix):
                print(f"Invalid sale: product lacks '{prefix}' prefix: {item}")
                continue
            sale = Sale(**item)
            valid_sales.append(sale)
        except ValidationError as e:
            print(f"Invalid sale: {item}, error: {e}")
            continue
    print(f"Valid sales: {len(valid_sales)}")
    return valid_sales

# Test
if __name__ == "__main__":
    data = [
        {"product": "Halal Mouse", "price": 24.99, "quantity": 10},
        {"product": "Monitor", "price": 199.99, "quantity": 2}
    ]
    valid_sales = validate_sales(data, "Halal")

import psycopg2

def connect_and_query() -> int:
    """Query sales count and close connection."""
    conn = psycopg2.connect(dbname="onboarding", user="postgres", password="", host="localhost", port="5432")
    try:
        cursor = conn.cursor()
        cursor.execute("SELECT COUNT(*) FROM sales")
        count = cursor.fetchone()[0]
        cursor.close()
        print(f"Sales count: {count}")
        return count
    finally:
        conn.close()
        print("Connection closed")

# Test
if __name__ == "__main__":
    count = connect_and_query()

import psycopg2
import sqlite3
import time
from pydantic import BaseModel, PositiveFloat, PositiveInt

class Sale(BaseModel):
    product: str
    price: PositiveFloat
    quantity: PositiveInt

def compare_db_performance(pg_conn: psycopg2.extensions.connection, sqlite_db: str, sale: Sale, output_path: str) -> None:
    """Insert sale into PostgreSQL, compare query performance, and save analysis."""
    results = []

    # Insert into PostgreSQL
    try:
        cursor = pg_conn.cursor()
        cursor.execute(
            "INSERT INTO sales (product, price, quantity) VALUES (%s, %s, %s)",
            (sale.product, sale.price, sale.quantity)
        )
        pg_conn.commit()
        cursor.close()
        results.append(f"Inserted sale: {sale.product}")
    except psycopg2.Error as e:
        results.append(f"PostgreSQL insert failed: {e}")
        with open(output_path, "w") as file:
            file.write("\n".join(results))
        return

    # Query PostgreSQL: Total sales
    pg_total_time = 0.0
    try:
        start_time = time.time()
        cursor = pg_conn.cursor()
        cursor.execute("SELECT SUM(price * quantity) FROM sales")
        pg_total = cursor.fetchone()[0] or 0.0
        pg_total_time = time.time() - start_time
        cursor.close()
        results.append(f"PostgreSQL total query time: {pg_total_time:.3f}s")
    except psycopg2.Error as e:
        results.append(f"PostgreSQL total query failed: {e}")

    # Query PostgreSQL: Grouped sales
    pg_grouped_time = 0.0
    try:
        start_time = time.time()
        cursor = pg_conn.cursor()
        cursor.execute("SELECT product, SUM(price * quantity) FROM sales GROUP BY product")
        pg_grouped = cursor.fetchall()
        pg_grouped_time = time.time() - start_time
        cursor.close()
        results.append(f"PostgreSQL grouped query time: {pg_grouped_time:.3f}s")
    except psycopg2.Error as e:
        results.append(f"PostgreSQL grouped query failed: {e}")

    # Query SQLite: Total sales
    sqlite_total_time = 0.0
    try:
        sqlite_conn = sqlite3.connect(sqlite_db)
        start_time = time.time()
        cursor = sqlite_conn.cursor()
        cursor.execute("SELECT SUM(price * quantity) FROM sales")
        sqlite_total = cursor.fetchone()[0] or 0.0
        sqlite_total_time = time.time() - start_time
        cursor.close()
        results.append(f"SQLite total query time: {sqlite_total_time:.3f}s")
    except sqlite3.Error as e:
        results.append(f"SQLite connection failed: {e}")

    # Query SQLite: Grouped sales
    sqlite_grouped_time = 0.0
    try:
        start_time = time.time()
        cursor = sqlite_conn.cursor()
        cursor.execute("SELECT product, SUM(price * quantity) FROM sales GROUP BY product")
        sqlite_grouped = cursor.fetchall()
        sqlite_grouped_time = time.time() - start_time
        cursor.close()
        results.append(f"SQLite grouped query time: {sqlite_grouped_time:.3f}s")
    except sqlite3.Error as e:
        results.append(f"SQLite grouped query failed: {e}")
    finally:
        sqlite_conn.close()

    # Save results
    explanation = (
        "\nExplanation: PostgreSQL is preferred over SQLite for Hijra Group’s analytics because it supports "
        "concurrent connections, advanced querying (e.g., CTEs), and enterprise features like replication, "
        "making it scalable for large transaction datasets. SQLite, while simple, is limited to single-user "
        "access and lacks these features. Grouped queries benefit from PostgreSQL’s optimization for complex analytics."
    )
    with open(output_path, "w") as file:
        file.write("\n".join(results) + explanation)
    for line in results:
        print(line)
    print(f"Results saved to {output_path}")

# Test
if __name__ == "__main__":
    pg_conn = connect_db("onboarding", "postgres", "", "localhost", "5432")
    sale = Sale(product="Halal Keyboard", price=49.99, quantity=5)
    compare_db_performance(pg_conn, "data/sales.db", sale, "ex6_concepts.txt")
    pg_conn.close()