refactor: replace allowed_column with required_columns and allow extra raw columns

Author: BLMgithub

.coveragerc

@@ -2,4 +2,5 @@
 omit =
     data_pipeline/prototypes/*
     data_pipeline/shared/table_configs.py
+    data_pipeline/stages/*
     */__init__.py

data_pipeline/shared/table_configs.py

@@ -10,7 +10,7 @@
     "df_orders": {
         "role": "event_fact",
         "primary_key": ["order_id"],
-        "allowed_column": [
+        "required_column": [
             "order_id",
             "customer_id",
             "order_status",
@@ -23,45 +23,36 @@
     "df_order_items": {
         "role": "transaction_detail",
         "primary_key": ["order_id"],
-        "allowed_column": [
+        "required_column": [
             "order_id",
             "product_id",
             "seller_id",
             "price",
-            "shipping_charges",
         ],
     },
     "df_customers": {
         "role": "entity_reference",
         "primary_key": ["customer_id"],
-        "allowed_column": [
+        "required_column": [
             "customer_id",
-            "customer_zip_code_prefix",
-            "customer_city",
             "customer_state",
         ],
     },
     "df_payments": {
         "role": "transaction_detail",
         "primary_key": ["order_id", "payment_sequential"],
-        "allowed_column": [
+        "required_column": [
             "order_id",
-            "payment_sequential",
-            "payment_type",
-            "payment_installments",
             "payment_value",
         ],
     },
     "df_products": {
         "role": "entity_reference",
         "primary_key": ["product_id"],
-        "allowed_column": [
+        "required_column": [
             "product_id",
             "product_category_name",
             "product_weight_g",
-            "product_length_cm",
-            "product_height_cm",
-            "product_width_cm",
         ],
     },
 }

data_pipeline/stages/assemble_validated_events.py

@@ -169,11 +169,6 @@ def assemble_events(run_context: RunContext) -> Dict:
 
     Grain:
     - One row per order_id (hard fail on violation)
-
-    Non-responsibilities:
-    - No validation
-    - No repair
-    - No business logic
     """
 
     report = init_report()

data_pipeline/stages/validate_raw_data.py

@@ -56,7 +56,7 @@ def run_base_validations(
     df: pd.DataFrame,
     table_name: str,
     primary_key: List[str],
-    allowed_column: List[str],
+    required_column: List[str],
     report: Dict[str, List[str]],
 ) -> bool:
     """
@@ -66,14 +66,13 @@ def run_base_validations(
     - `errors` - dataset is structurally invalid; downstream validations should stop
     - `warnings` - admissible data quality issues that may be repairable
 
-    **errors:**
+    errors:
     - dataset is empty
-    - missing allowed column(s)
-    - unexpected extra column(s)
+    - missing required column(s)
     - missing primary key column(s)
     - conflicting duplicate primary keys
 
-    **warnings:**
+    warnings:
     - duplicate columns
     - null primary key values
     - identical duplicates
@@ -85,19 +84,15 @@ def run_base_validations(
         return False
 
     actual = set(df.columns)
-    allowed = set(allowed_column)
+    required = set(required_column)
 
-    missing_allowed = sorted(allowed - actual)
-    if missing_allowed:
-        log_error(f"{table_name}: missing allowed column(s): {missing_allowed}", report)
-
-    invalid_column = sorted(actual - allowed)
-    if invalid_column:
+    missing_required = sorted(required - actual)
+    if missing_required:
         log_error(
-            f"{table_name}: non-allowed extra column(s): {invalid_column}", report
+            f"{table_name}: missing required column(s): {missing_required}", report
         )
 
-    if missing_allowed or invalid_column:
+    if missing_required:
         return False
 
     missing_pk_columns = [col for col in primary_key if col not in df.columns]
@@ -174,10 +169,10 @@ def run_event_fact_validations(
     - `errors` - dataset is structurally invalid; downstream validations should stop
     - `warnings` - admissible data quality issues that may be repairable
 
-    **errors:**
+    errors:
     - missing required timestamp column(s)
 
-    **warnings:**
+    warnings:
     - unparsable timestamp values in required timestamp fields
     - approval timestamp earlier than purchase timestamp
     - delivery timestamp earlier than purchase timestamp
@@ -246,7 +241,7 @@ def run_transaction_detail_validations(
 
     Collects error-level data quality findings.
 
-    **errors:**
+    errors:
     - negative values in numeric columns
     """
 
@@ -278,10 +273,10 @@ def run_cross_table_validations(
     - `warnings` - admissible referential integrity issues
     - `info` - validation skipped due to missing required tables
 
-    **info:**
+    info:
     - validation skipped when required tables are unavailable
 
-    **warnings:**
+    warnings:
     - order items referencing non-existent order_id
     - payments referencing non-existent order_id
     """
@@ -341,10 +336,6 @@ def apply_validation(run_context: RunContext, base_path: Path | None = None) ->
     Severity model:
     - errors: structurally invalid → halt upstream
     - warnings: admissible but repairable issues
-
-    Non-responsibilities:
-    - No data mutation
-    - No process termination
     """
 
     if base_path is None:
@@ -374,7 +365,7 @@ def error(msg: str):
         tables[table_name] = df
 
         if not run_base_validations(
-            df, table_name, config["primary_key"], config["allowed_column"], report
+            df, table_name, config["primary_key"], config["required_column"], report
         ):
             continue