41 - Type-Safe Data Processing

# File: de-onboarding/sales_model.py
from pydantic import BaseModel, field_validator  # Import Pydantic
from typing import List  # For type annotations

class Sale(BaseModel):  # Define Sale model
    product: str  # Product name (string)
    price: float  # Price (float)
    quantity: int  # Quantity (integer)

    @field_validator("product")  # Validate product
    @classmethod
    def check_product(cls, value: str) -> str:
        """Ensure product starts with 'Halal'."""
        if not value.startswith("Halal"):  # Check prefix
            raise ValueError(f"Product must start with 'Halal': {value}")  # Raise error
        return value

    @field_validator("price")  # Validate price
    @classmethod
    def check_price(cls, value: float) -> float:
        """Ensure price is positive and has <= 2 decimals."""
        if value <= 0:  # Check positive
            raise ValueError(f"Price must be positive: {value}")  # Raise error
        if round(value, 2) != value:  # Check decimals
            raise ValueError(f"Price must have <= 2 decimals: {value}")  # Raise error
        return value

    @field_validator("quantity")  # Validate quantity
    @classmethod
    def check_quantity(cls, value: int) -> int:
        """Ensure quantity is positive and <= 100."""
        if value <= 0 or value > 100:  # Check range
            raise ValueError(f"Quantity must be 1–100: {value}")  # Raise error
        return value

class SalesConfig(BaseModel):  # Define config model
    min_price: float  # Minimum price
    max_quantity: int  # Maximum quantity
    required_fields: List[str]  # Required fields
    product_prefix: str  # Product prefix
    max_decimals: int  # Maximum decimals

# File: de-onboarding/validate_sales.py
from sales_model import Sale  # Import Sale model
import pandas as pd  # For DataFrame
from typing import List, Dict, Any  # For type annotations
import logging  # For logging

# Configure logging
logging.basicConfig(
    level=logging.INFO,  # Log INFO and above
    format="%(asctime)s - %(levelname)s - %(message)s"  # Log format
)
logger = logging.getLogger(__name__)  # Create logger

def validate_sales_data(df: pd.DataFrame) -> List[Dict[str, Any]]:
    """Validate sales data using Pydantic."""
    validated_sales: List[Dict[str, Any]] = []  # Store validated sales
    for index, row in df.iterrows():  # Iterate rows
        try:
            sale = Sale(
                product=str(row["product"]),  # Convert to string
                price=float(row["price"]),  # Convert to float
                quantity=int(row["quantity"])  # Convert to int
            )
            validated_sales.append(sale.dict())  # Add validated sale
            logger.info(f"Validated sale: {sale}")  # Log success
        except (ValueError, TypeError) as e:
            logger.error(f"Invalid sale at index {index}: {row.to_dict()} - {str(e)}")  # Log error
            continue
    return validated_sales  # Return validated sales

# Test
if __name__ == "__main__":
    df = pd.DataFrame({
        "product": ["Halal Laptop", "Mouse", "Halal Mouse"],
        "price": [999.99, -10.0, 24.99],
        "quantity": [2, 5, 10]
    })
    sales = validate_sales_data(df)
    print("Validated Sales:", sales)

{
  "include": ["de-onboarding"],
  "exclude": ["**/venv"],
  "strict": ["de-onboarding/*.py"],
  "typeCheckingMode": "strict",
  "reportMissingTypeStubs": false
}

from pydantic import validate_arguments
from typing import List
from sales_model import Sale

@validate_arguments
def validate_batch(sales: List[Dict[str, Any]]) -> List[Sale]:
    """Validate a batch of sales."""
    return [Sale(**sale) for sale in sales]

# File: de-onboarding/process_sales.py
from sales_model import Sale, SalesConfig  # Import models
import pandas as pd  # For DataFrame
import numpy as np  # For computations
import logging  # For logging
from typing import Tuple, Dict, Any  # For type annotations

logger = logging.getLogger(__name__)  # Create logger

def process_sales_data(df: pd.DataFrame, config: SalesConfig) -> Tuple[Dict[str, Any], int]:
    """Process sales data with type-safe validation."""
    validated_sales = []  # Store validated sales
    for index, row in df.iterrows():  # Iterate rows
        try:
            sale = Sale(
                product=str(row["product"]),
                price=float(row["price"]),
                quantity=int(row["quantity"])
            )
            if sale.price < config.min_price or sale.quantity > config.max_quantity:
                logger.error(f"Sale violates config: {sale}")
                continue
            validated_sales.append(sale)
            logger.info(f"Processed sale: {sale}")
        except (ValueError, TypeError) as e:
            logger.error(f"Invalid sale at index {index}: {row.to_dict()} - {str(e)}")
            continue

    if not validated_sales:  # Check if empty
        logger.warning("No valid sales data")
        return {"total_sales": 0.0, "unique_products": [], "top_products": {}}, 0

    # Convert to DataFrame
    df_valid = pd.DataFrame([sale.dict() for sale in validated_sales])
    df_valid["amount"] = df_valid["price"] * df_valid["quantity"]

    # Compute metrics
    total_sales = float(np.sum(df_valid["amount"].values))  # Total sales
    unique_products = df_valid["product"].unique().tolist()  # Unique products
    sales_by_product = df_valid.groupby("product")["amount"].sum()
    top_products = sales_by_product.sort_values(ascending=False).head(3).to_dict()

    return {
        "total_sales": total_sales,
        "unique_products": unique_products,
        "top_products": top_products
    }, len(df_valid)

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

# File: de-onboarding/sales_model.py (same as above)

# File: de-onboarding/sales_processor.py
from sales_model import Sale, SalesConfig
import pandas as pd
import numpy as np
import yaml
import json
import logging
from typing import Tuple, Dict, Any
import os

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler("data/sales.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

def read_config(config_path: str) -> SalesConfig:
    """Read and validate YAML configuration."""
    logger.info(f"Opening config: {config_path}")
    with open(config_path, "r", encoding="utf-8") as file:
        config_data = yaml.safe_load(file)
    config = SalesConfig(**config_data)
    logger.info(f"Loaded config: {config}")
    return config

def load_and_validate_sales(csv_path: str, config: SalesConfig) -> Tuple[pd.DataFrame, int, int]:
    """Load and validate sales CSV."""
    logger.info(f"Loading CSV: {csv_path}")
    df = pd.read_csv(csv_path, encoding="utf-8")
    logger.info(f"Initial DataFrame:\n{df.head()}")

    required_fields = config.required_fields
    missing_fields = [f for f in required_fields if f not in df.columns]
    if missing_fields:
        logger.error(f"Missing columns: {missing_fields}")
        return pd.DataFrame(), 0, len(df)

    validated_sales = []
    for index, row in df.iterrows():
        try:
            sale = Sale(
                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"]) else 0
            )
            if sale.price < config.min_price or sale.quantity > config.max_quantity:
                logger.error(f"Sale violates config: {sale}")
                continue
            validated_sales.append(sale)
            logger.info(f"Validated sale: {sale}")
        except (ValueError, TypeError) as e:
            logger.error(f"Invalid sale at index {index}: {row.to_dict()} - {str(e)}")
            continue

    df_valid = pd.DataFrame([sale.dict() for sale in validated_sales])
    return df_valid, len(df_valid), len(df)

def process_sales(df: pd.DataFrame, config: SalesConfig) -> Tuple[Dict[str, Any], int]:
    """Process sales data."""
    if df.empty:
        logger.warning("No valid sales data")
        return {"total_sales": 0.0, "unique_products": [], "top_products": {}}, 0

    df["amount"] = df["price"] * df["quantity"]
    logger.info(f"DataFrame with Amount:\n{df}")

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

    valid_sales = len(df)
    logger.info(f"Valid sales: {valid_sales} records")

    return {
        "total_sales": total_sales,
        "unique_products": unique_products,
        "top_products": top_products
    }, valid_sales

def export_results(results: Dict[str, Any], json_path: str) -> None:
    """Export results to JSON."""
    logger.info(f"Writing to: {json_path}")
    with open(json_path, "w", encoding="utf-8") as file:
        json.dump(results, file, indent=2)
    logger.info(f"Exported results to {json_path}")

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

    config = read_config(config_path)
    df, valid_sales, total_records = load_and_validate_sales(csv_path, config)
    results, valid_sales = process_sales(df, config)
    export_results(results, json_path)

    logger.info("\nSales Report:")
    logger.info(f"Total Records Processed: {total_records}")
    logger.info(f"Valid Sales: {valid_sales}")
    logger.info(f"Invalid Sales: {total_records - valid_sales}")
    logger.info(f"Total Sales: ${round(results['total_sales'], 2)}")
    logger.info(f"Unique Products: {results['unique_products']}")
    logger.info(f"Top Products: {results['top_products']}")

if __name__ == "__main__":
    main()

# File: de-onboarding/tests/test_sales_processor.py
from sales_processor import read_config, load_and_validate_sales, process_sales
from sales_model import SalesConfig
import pandas as pd
import pytest
import os

@pytest.fixture
def config():
    """Fixture for SalesConfig."""
    return SalesConfig(
        min_price=10.0,
        max_quantity=100,
        required_fields=["product", "price", "quantity"],
        product_prefix="Halal",
        max_decimals=2
    )

def test_read_config(config):
    """Test reading config."""
    config_path = "data/config.yaml"
    loaded_config = read_config(config_path)
    assert loaded_config.min_price == config.min_price
    assert loaded_config.max_quantity == config.max_quantity

def test_valid_sales(config):
    """Test valid sales data."""
    df = pd.DataFrame({
        "product": ["Halal Laptop", "Halal Mouse"],
        "price": [999.99, 24.99],
        "quantity": [2, 10]
    })
    df_valid, valid_sales, total_records = load_and_validate_sales("data/sales.csv", config)
    results, count = process_sales(df_valid, config)
    assert count == 3
    assert results["total_sales"] == pytest.approx(2499.83, rel=1e-2)
    assert "Halal Laptop" in results["unique_products"]

def test_invalid_sales(config):
    """Test invalid sales data."""
    df = pd.DataFrame({
        "product": ["Mouse"],
        "price": [-10.0],
        "quantity": [5]
    })
    df_valid, valid_sales, total_records = load_and_validate_sales("data/sales.csv", config)
    assert valid_sales <= total_records

{
  "total_sales": 2499.83,
  "unique_products": ["Halal Laptop", "Halal Mouse", "Halal Keyboard"],
  "top_products": {
    "Halal Laptop": 1999.98,
    "Halal Mouse": 249.9,
    "Halal Keyboard": 249.95
  }
}

sale = Sale(product="Halal Laptop", price=9.99, quantity=2)  # Raises ValueError

def process(df: pd.DataFrame) -> dict:  # Type error: dict is untyped
    return {"total": sum(df["price"] * df["quantity"])}

from pydantic import BaseModel, field_validator, ValidationError
from typing import ClassVar

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

    @field_validator("price")
    @classmethod
    def check_price(cls, value: float) -> float:
        if value < 10.0:
            raise ValueError(f"Price must be >= 10.0: {value}")
        if value <= 0:
            raise ValueError(f"Price must be positive: {value}")
        if round(value, 2) != value:
            raise ValueError(f"Price must have <= 2 decimals: {value}")
        return value

    @field_validator("product")
    @classmethod
    def check_product(cls, value: str) -> str:
        if not value.startswith("Halal"):
            raise ValueError(f"Product must start with 'Halal': {value}")
        return value

    @field_validator("quantity")
    @classmethod
    def check_quantity(cls, value: int) -> int:
        if value <= 0 or value > 100:
            raise ValueError(f"Quantity must be 1–100: {value}")
        return value

# Test
if __name__ == "__main__":
    try:
        sale = Sale(product="Halal Laptop", price=9.99, quantity=2)
    except ValidationError as e:
        print(f"Error: {e.errors()}")

from sales_model import Sale
import pandas as pd
import logging
from typing import List, Dict, Any

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def validate_df(df: pd.DataFrame) -> List[Dict[str, Any]]:
    """Validate DataFrame using Pydantic."""
    logger.info(f"DataFrame types: {df.dtypes}")  # Debug types
    validated = []
    for index, row in df.iterrows():
        try:
            sale = Sale(
                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"]) else 0
            )
            validated.append(sale.dict())
            logger.info(f"Validated: {sale}")
        except ValueError as e:
            logger.error(f"Invalid at index {index}: {str(e)}")
    return validated

# Test
if __name__ == "__main__":
    df = pd.DataFrame({
        "product": ["Halal Laptop", "Mouse"],
        "price": [999.99, -10.0],
        "quantity": [2, 5]
    })
    print("Validated Sales:", validate_df(df))

from sales_processor import process_sales
from sales_model import SalesConfig
import pandas as pd
import pytest

def test_process_sales():
    """Test process_sales function."""
    df = pd.DataFrame({
        "product": ["Halal Laptop"],
        "price": [999.99],
        "quantity": [2]
    })
    config = SalesConfig(
        min_price=10.0, max_quantity=100, required_fields=["product", "price", "quantity"],
        product_prefix="Halal", max_decimals=2
    )
    results, count = process_sales(df, config)
    assert results["total_sales"] == 1999.98
    assert count == 1
    # Test empty DataFrame
    empty_df = pd.DataFrame(columns=["product", "price", "quantity"])
    results, count = process_sales(empty_df, config)
    assert results["total_sales"] == 0.0
    assert count == 0

# Run with: pytest ex3_test.py -v

from typing import Dict, Any
import pandas as pd

def process(df: pd.DataFrame) -> Dict[str, Any]:
    """Process DataFrame and compute total sales."""
    return {"total": float(sum(df["price"] * df["quantity"]))}

# Test
if __name__ == "__main__":
    df = pd.DataFrame({
        "product": ["Halal Laptop", "Halal Mouse"],
        "price": [999.99, 24.99],
        "quantity": [2, 10]
    })
    print(process(df))  # Expected: {'total': 2249.88}

from sales_model import Sale
import pandas as pd
import time
import logging
import sys
from typing import List, Dict, Any

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def manual_validate(df: pd.DataFrame) -> List[Dict[str, Any]]:
    """Manually validate sales data."""
    validated = []
    for index, row in df.iterrows():
        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"]) else 0
        if (product.startswith("Halal") and price > 0 and 0 < quantity <= 100 and
                round(price, 2) == price):
            validated.append({"product": product, "price": price, "quantity": quantity})
        else:
            logger.error(f"Manual validation failed at index {index}")
    return validated

def pydantic_validate(df: pd.DataFrame) -> List[Dict[str, Any]]:
    """Validate sales data with Pydantic."""
    validated = []
    for index, row in df.iterrows():
        try:
            sale = Sale(
                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"]) else 0
            )
            validated.append(sale.dict())
        except ValueError as e:
            logger.error(f"Pydantic validation failed at index {index}: {str(e)}")
    return validated

def compare_validations(csv_path: str, comparison_path: str, reflection_path: str) -> None:
    """Compare Pydantic and manual validation speeds."""
    df = pd.read_csv(csv_path, encoding="utf-8")

    # Time Pydantic validation (average 5 runs)
    pydantic_times = []
    for _ in range(5):
        start_time = time.time()
        pydantic_valid = pydantic_validate(df)
        pydantic_times.append(time.time() - start_time)
    pydantic_time = sum(pydantic_times) / len(pydantic_times)

    # Time manual validation (average 5 runs)
    manual_times = []
    for _ in range(5):
        start_time = time.time()
        manual_valid = manual_validate(df)
        manual_times.append(time.time() - start_time)
    manual_time = sum(manual_times) / len(manual_times)

    # Measure memory usage
    pydantic_memory = sys.getsizeof(pydantic_valid)
    manual_memory = sys.getsizeof(manual_valid)

    # Write comparison results
    explanation = (
        f"Pydantic Time: {pydantic_time:.3f}s\n"
        f"Manual Time: {manual_time:.3f}s\n"
        f"Pydantic Memory: {pydantic_memory} bytes\n"
        f"Manual Memory: {manual_memory} bytes\n"
        f"Explanation: Pydantic ensures type safety and compliance for Hijra Group’s "
        f"analytics, catching errors early, but manual validation may be faster for simple rules."
    )
    with open(comparison_path, "w", encoding="utf-8") as f:
        f.write(explanation)
    logger.info(f"Comparison written to {comparison_path}")

    # Write reflection
    reflection = (
        "Timing differences may occur due to CPU speed, memory availability, "
        "or system load across machines."
    )
    with open(reflection_path, "w", encoding="utf-8") as f:
        f.write(reflection)
    logger.info(f"Reflection written to {reflection_path}")

# Test
if __name__ == "__main__":
    compare_validations("data/sales.csv", "data/validation_comparison.txt", "data/validation_reflection.txt")