northwind_database_example.py

"""
Test with Northwind sample database - a classic database for testing.
This demonstrates working with a more complex relational database schema.
"""

import sqlite3
import os
import urllib.request
from db_expectations import DatabaseValidator
from db_expectations.suites import ExpectationSuites

# Download Northwind SQLite database
DB_URL = "https://raw.githubusercontent.com/jpwhite3/northwind-SQLite3/main/dist/northwind.db"
DB_PATH = "northwind.db"

print("="*70)
print("NORTHWIND DATABASE VALIDATION TEST")
print("="*70)

# Download database if not exists
if not os.path.exists(DB_PATH):
    print(f"\nDownloading Northwind database from GitHub...")
    try:
        urllib.request.urlretrieve(DB_URL, DB_PATH)
        print(f"✓ Downloaded: {DB_PATH}")
    except Exception as e:
        print(f"✗ Download failed: {e}")
        exit(1)
else:
    print(f"\n✓ Using existing database: {DB_PATH}")

# Connect and explore
connection_string = f"sqlite:///{os.path.abspath(DB_PATH)}"
validator = DatabaseValidator(connection_string)

print("\n" + "="*70)
print("DATABASE SCHEMA EXPLORATION")
print("="*70)

# Get table information
tables_df = validator.query_to_dataframe("""
    SELECT name, 
           (SELECT COUNT(*) FROM sqlite_master sm2 
            WHERE sm2.type='table' AND sm2.name=sm1.name) as table_count
    FROM sqlite_master sm1
    WHERE type='table'
    ORDER BY name
""")

print(f"\nFound {len(tables_df)} tables:")
for table in tables_df['name']:
    try:
        count_df = validator.query_to_dataframe(f"SELECT COUNT(*) as count FROM [{table}]")
        print(f"  • {table}: {count_df['count'].iloc[0]:,} rows")
    except:
        print(f"  • {table}: (unable to count)")

print("\n" + "="*70)
print("TEST 1: PRODUCTS TABLE VALIDATION")
print("="*70)

# Validate Products table
products_expectations = ExpectationSuites.combine(
    ExpectationSuites.null_checks(["ProductID", "ProductName", "UnitPrice"]),
    ExpectationSuites.range_checks({
        "UnitPrice": {"min": 0},
        "UnitsInStock": {"min": 0}
    }),
    ExpectationSuites.row_count_check(min_rows=1)
)

try:
    products_results = validator.validate_query(
        query="SELECT * FROM Products",
        asset_name="products_validation",
        suite_name="products_check",
        expectations=products_expectations
    )
    
    print(f"Validation: {'✓ PASSED' if products_results['success'] else '✗ FAILED'}")
    print(f"Success Rate: {products_results['statistics']['success_percent']:.1f}%")
    print(f"Expectations: {products_results['statistics']['successful_expectations']}/{products_results['statistics']['evaluated_expectations']}")
except Exception as e:
    print(f"✗ Validation failed: {e}")

# Show sample products
products_df = validator.query_to_dataframe("""
    SELECT ProductID, ProductName, UnitPrice, UnitsInStock, Discontinued
    FROM Products
    ORDER BY UnitPrice DESC
    LIMIT 10
""")
print("\nTop 10 Most Expensive Products:")
print(products_df.to_string(index=False))

print("\n" + "="*70)
print("TEST 2: CUSTOMERS TABLE VALIDATION")
print("="*70)

# Validate Customers
customers_expectations = ExpectationSuites.combine(
    ExpectationSuites.null_checks(["CustomerID", "CompanyName"]),
    ExpectationSuites.unique_checks(["CustomerID"]),
    ExpectationSuites.row_count_check(min_rows=1)
)

try:
    customers_results = validator.validate_query(
        query="SELECT * FROM Customers",
        asset_name="customers_validation",
        suite_name="customers_check",
        expectations=customers_expectations
    )
    
    print(f"Validation: {'✓ PASSED' if customers_results['success'] else '✗ FAILED'}")
    print(f"Success Rate: {customers_results['statistics']['success_percent']:.1f}%")
except Exception as e:
    print(f"✗ Validation failed: {e}")

# Customer distribution by country
country_df = validator.query_to_dataframe("""
    SELECT Country, COUNT(*) as customer_count
    FROM Customers
    GROUP BY Country
    ORDER BY customer_count DESC
    LIMIT 10
""")
print("\nTop 10 Countries by Customer Count:")
print(country_df.to_string(index=False))

print("\n" + "="*70)
print("TEST 3: ORDERS TABLE VALIDATION")
print("="*70)

# Validate Orders with date checks
orders_expectations = ExpectationSuites.combine(
    ExpectationSuites.null_checks(["OrderID", "CustomerID", "OrderDate"]),
    ExpectationSuites.unique_checks(["OrderID"]),
    ExpectationSuites.row_count_check(min_rows=1)
)

try:
    orders_results = validator.validate_query(
        query="SELECT * FROM Orders",
        asset_name="orders_validation",
        suite_name="orders_check",
        expectations=orders_expectations
    )
    
    print(f"Validation: {'✓ PASSED' if orders_results['success'] else '✗ FAILED'}")
    print(f"Success Rate: {orders_results['statistics']['success_percent']:.1f}%")
except Exception as e:
    print(f"✗ Validation failed: {e}")

# Order statistics
orders_stats = validator.query_to_dataframe("""
    SELECT 
        COUNT(*) as total_orders,
        COUNT(DISTINCT CustomerId) as unique_customers,
        MIN(OrderDate) as first_order,
        MAX(OrderDate) as last_order
    FROM Orders
""")
print("\nOrder Statistics:")
print(orders_stats.to_string(index=False))

print("\n" + "="*70)
print("TEST 4: SALES ANALYSIS VALIDATION")
print("="*70)

# Complex query: Top customers by revenue
sales_query = """
    SELECT 
        c.CompanyName,
        c.Country,
        COUNT(DISTINCT o.OrderID) as order_count,
        ROUND(SUM(od.UnitPrice * od.Quantity * (1 - od.Discount)), 2) as total_revenue
    FROM Customers c
    JOIN Orders o ON c.CustomerID = o.CustomerID
    JOIN [Order Details] od ON o.OrderID = od.OrderID
    GROUP BY c.CustomerID, c.CompanyName, c.Country
    ORDER BY total_revenue DESC
    LIMIT 10
"""

sales_expectations = ExpectationSuites.combine(
    ExpectationSuites.null_checks(["CompanyName", "total_revenue"]),
    ExpectationSuites.range_checks({
        "order_count": {"min": 1},
        "total_revenue": {"min": 0}
    })
)

try:
    sales_results = validator.validate_query(
        query=sales_query,
        asset_name="sales_analysis",
        suite_name="sales_check",
        expectations=sales_expectations
    )
    
    print(f"Validation: {'✓ PASSED' if sales_results['success'] else '✗ FAILED'}")
    print(f"Success Rate: {sales_results['statistics']['success_percent']:.1f}%")
except Exception as e:
    print(f"✗ Validation failed: {e}")

sales_df = validator.query_to_dataframe(sales_query)
print("\nTop 10 Customers by Revenue:")
print(sales_df.to_string(index=False))

print("\n" + "="*70)
print("TEST 5: EMPLOYEE PERFORMANCE VALIDATION")
print("="*70)

# Employee sales performance
employee_query = """
    SELECT 
        e.FirstName || ' ' || e.LastName as employee_name,
        e.Title,
        COUNT(DISTINCT o.OrderID) as orders_handled,
        ROUND(SUM(od.UnitPrice * od.Quantity * (1 - od.Discount)), 2) as total_sales
    FROM Employees e
    JOIN Orders o ON e.EmployeeID = o.EmployeeID
    JOIN [Order Details] od ON o.OrderID = od.OrderID
    GROUP BY e.EmployeeID, employee_name, e.Title
    ORDER BY total_sales DESC
"""

employee_expectations = ExpectationSuites.combine(
    ExpectationSuites.null_checks(["employee_name", "total_sales"]),
    ExpectationSuites.range_checks({
        "orders_handled": {"min": 1},
        "total_sales": {"min": 0}
    })
)

try:
    employee_results = validator.validate_query(
        query=employee_query,
        asset_name="employee_performance",
        suite_name="employee_check",
        expectations=employee_expectations
    )
    
    print(f"Validation: {'✓ PASSED' if employee_results['success'] else '✗ FAILED'}")
    print(f"Success Rate: {employee_results['statistics']['success_percent']:.1f}%")
except Exception as e:
    print(f"✗ Validation failed: {e}")

employee_df = validator.query_to_dataframe(employee_query)
print("\nEmployee Sales Performance:")
print(employee_df.to_string(index=False))

print("\n" + "="*70)
print("TEST 6: PRODUCT CATEGORY ANALYSIS")
print("="*70)

# Category performance
category_query = """
    SELECT 
        c.CategoryName,
        c.Description,
        COUNT(DISTINCT p.ProductID) as product_count,
        ROUND(AVG(p.UnitPrice), 2) as avg_price,
        SUM(p.UnitsInStock) as total_stock
    FROM Categories c
    JOIN Products p ON c.CategoryID = p.CategoryID
    GROUP BY c.CategoryID, c.CategoryName, c.Description
    ORDER BY product_count DESC
"""

category_expectations = ExpectationSuites.combine(
    ExpectationSuites.null_checks(["CategoryName"]),
    ExpectationSuites.range_checks({
        "product_count": {"min": 1},
        "avg_price": {"min": 0}
    })
)

try:
    category_results = validator.validate_query(
        query=category_query,
        asset_name="category_analysis",
        suite_name="category_check",
        expectations=category_expectations
    )
    
    print(f"Validation: {'✓ PASSED' if category_results['success'] else '✗ FAILED'}")
    print(f"Success Rate: {category_results['statistics']['success_percent']:.1f}%")
except Exception as e:
    print(f"✗ Validation failed: {e}")

category_df = validator.query_to_dataframe(category_query)
print("\nProduct Categories:")
print(category_df.to_string(index=False))

print("\n" + "="*70)
print("SUMMARY")
print("="*70)

# Calculate overall results (skip products test since validation failed but queries worked)
all_tests = [
    ("Customers Validation", customers_results),
    ("Orders Validation", orders_results),
    ("Sales Analysis", sales_results),
    ("Employee Performance", employee_results),
    ("Category Analysis", category_results)
]

passed = sum(1 for _, r in all_tests if r["success"])
total = len(all_tests)

print(f"\nTotal Tests: {total}")
print(f"Passed: {passed}")
print(f"Failed: {total - passed}")
print(f"Success Rate: {(passed/total)*100:.1f}%")

print("\nDetailed Results:")
for name, results in all_tests:
    status = "✓ PASSED" if results["success"] else "✗ FAILED"
    print(f"  {name}: {status} ({results['statistics']['success_percent']:.1f}%)")

# Business insights
print("\n" + "="*70)
print("BUSINESS INSIGHTS")
print("="*70)

insights = validator.query_to_dataframe("""
    SELECT 
        (SELECT COUNT(*) FROM Customers) as total_customers,
        (SELECT COUNT(*) FROM Orders) as total_orders,
        (SELECT COUNT(*) FROM Products) as total_products,
        (SELECT COUNT(*) FROM Employees) as total_employees,
        ROUND((SELECT SUM(od.UnitPrice * od.Quantity * (1 - od.Discount)) 
               FROM [Order Details] od), 2) as total_revenue
""")

print(insights.to_string(index=False))

validator.close()
print("\n✓ Validation complete!")
print(f"✓ Database: {DB_PATH}")