How to Validate Fact Table Grain Row Counts in Python

import csv
import hashlib
from pathlib import Path


def validate_fact_grain(fact_file: Path, expected_dim_keys: dict[str, set[str]]) -> dict:
    """
    Validate fact table grain by checking each row's dimension keys exist
    in expected dimension tables and row count consistency.
    """
    dim_references = {}
    with open(fact_file, newline="") as f:
        reader = csv.DictReader(f)
        rows = list(reader)
        for row in rows:
            for dim_name, key_col in expected_dim_keys.items():
                key_value = row.get(key_col)
                if key_value is None:
                    raise ValueError(f"Missing column {key_col}")
                dim_references.setdefault(dim_name, set()).add(key_value)

        # Check grain uniqueness based on all dimension key columns combined
        grain_key_cols = sorted(expected_dim_keys.values())
        seen_grains = set()
        duplicate_count = 0
        for row in rows:
            grain = tuple(row[col] for col in grain_key_cols)
            if grain in seen_grains:
                duplicate_count += 1
            else:
                seen_grains.add(grain)

    return {
        "row_count": len(rows),
        "unique_grain_count": len(seen_grains),
        "duplicate_grain_rows": duplicate_count,
        "dimension_cardinality": {dim: len(keys) for dim, keys in dim_references.items()},
    }


if __name__ == "__main__":
    fact_data = [
        {"order_id": "O1", "product_id": "P1", "qty": "2"},
        {"order_id": "O1", "product_id": "P2", "qty": "1"},
        {"order_id": "O2", "product_id": "P1", "qty": "5"},
        {"order_id": "O1", "product_id": "P1", "qty": "2"},  # duplicate grain
    ]
    fact_path = Path("/tmp/fact_sales.csv")
    with open(fact_path, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=["order_id", "product_id", "qty"])
        writer.writeheader()
        writer.writerows(fact_data)

    result = validate_fact_grain(
        fact_path,
        expected_dim_keys={"dim_order": "order_id", "dim_product": "product_id"},
    )
    print(result)