fixed pipeline failure

Author: gibahjoe

digital_land/expectations/operations/csv.py

@@ -40,16 +40,6 @@ def _sql_string(value) -> str:
     return f"'{cleaned}'"
 
 
-def _normalize_condition_groups(conditions, name: str) -> list:
-    if conditions is None:
-        return []
-    if isinstance(conditions, dict):
-        return [conditions]
-    if isinstance(conditions, list):
-        return conditions
-    raise ValueError(f"{name} must be a dict, list of dicts, or None")
-
-
 def _build_field_condition(field_name: str, spec) -> str:
     if isinstance(spec, dict):
         op = str(spec.get("op", spec.get("operation", ""))).strip().lower()
@@ -79,27 +69,35 @@ def _build_field_condition(field_name: str, spec) -> str:
     )
 
 
-def _build_condition_group(group: dict, file_columns: list) -> str:
-    if not isinstance(group, dict) or not group:
-        raise ValueError("Each condition group must be a non-empty dict")
-
-    parts = []
-    for field_name, spec in group.items():
-        if field_name not in file_columns:
-            raise ValueError(
-                f"Column '{field_name}' not found in file. Available columns: {file_columns}"
-            )
-        parts.append(_build_field_condition(field_name, spec))
-
-    return f"({' AND '.join(parts)})"
-
-
 def _build_filter_clause(filter_spec, file_columns: list, name: str) -> str:
     """Build SQL clause that keeps rows matching structured conditions."""
-    groups = _normalize_condition_groups(filter_spec, name)
+    if filter_spec is None:
+        groups = []
+    elif isinstance(filter_spec, dict):
+        groups = [filter_spec]
+    elif isinstance(filter_spec, list):
+        groups = filter_spec
+    else:
+        raise ValueError(f"{name} must be a dict, list of dicts, or None")
+
     if not groups:
         return ""
-    clauses = [_build_condition_group(group, file_columns) for group in groups]
+
+    clauses = []
+    for group in groups:
+        if not isinstance(group, dict) or not group:
+            raise ValueError("Each condition group must be a non-empty dict")
+
+        parts = []
+        for field_name, spec in group.items():
+            if field_name not in file_columns:
+                raise ValueError(
+                    f"Column '{field_name}' not found in file. Available columns: {file_columns}"
+                )
+            parts.append(_build_field_condition(field_name, spec))
+
+        clauses.append(f"({' AND '.join(parts)})")
+
     return f" AND ({' OR '.join(clauses)})"
 
 

tests/integration/expectations/checkpoints/test_csv.py

@@ -1,5 +1,4 @@
 import csv
-import json
 import pytest
 
 from digital_land.expectations.checkpoints.csv import CsvCheckpoint