Fix: Revert index type to str and handle empty DFs (#32)

Author: adrien-berchet

data_validation_framework/task.py

@@ -295,14 +295,10 @@ def kwargs(self):
     def read_dataset(self):
         """Import the dataset to a :class:`pandas.DataFrame`.
 
-        Note that the index column is loaded as a string.
-
         This method can be overridden to load custom data (e.g. GeoDataFrame, etc.).
         The dataset should always be loaded from the path given by `self.dataset_df`.
         """
-        return pd.read_csv(
-            self.dataset_df, index_col=self.input_index_col, dtype={self.input_index_col: str}
-        )
+        return pd.read_csv(self.dataset_df, index_col=self.input_index_col)
 
     def pre_process(self, df, args, kwargs):
         """Method executed before applying the external function."""

data_validation_framework/util.py

@@ -92,6 +92,8 @@ def message_worker(progress_bar, message_queue):
 
 def apply_to_df(df, func, *args, nb_processes=None, redirect_stdout=None, **kwargs):
     """Apply a function to df rows using tqdm."""
+    if df.empty:
+        return df
     nb_jobs = len(df)
     if redirect_stdout is None:
         redirect_stdout = True

tests/test_task.py

@@ -1288,9 +1288,7 @@ def check_exception(failed_task, exception):  # pylint: disable=unused-variable
         assert not luigi.build([failing_task], local_scheduler=True)
 
         assert failed_tasks == [str(failing_task)]
-        assert exceptions == [
-            str(IndexError("The following index values are duplicated: ['0', '1']"))
-        ]
+        assert exceptions == [str(IndexError("The following index values are duplicated: [0, 1]"))]
 
     def test_change_index(self, tmpdir, TestTask):
         """Test that the process fails if the index is changed by the validation function."""
@@ -1743,6 +1741,27 @@ def inputs(self):
             == redirect_stdout
         )
 
+    @pytest.fixture
+    def empty_dataset_df_path(self, tmpdir):
+        """Create an empty CSV dataset and return its path."""
+        dataset_df_path = tmpdir / "dataset.csv"
+        base_dataset_df = pd.DataFrame({"a": [], "b": []})
+        base_dataset_df.to_csv(dataset_df_path)
+
+        return str(dataset_df_path)
+
+    def test_defaults_with_empty(self, TestTask, empty_dataset_df_path, tmpdir):
+        """Test that the empty dataset is properly processed."""
+        assert luigi.build(
+            [TestTask(dataset_df=empty_dataset_df_path, result_path=str(tmpdir / "out"))],
+            local_scheduler=True,
+        )
+        result = pd.read_csv(tmpdir / "out" / "TestTask" / "report.csv")
+        assert result["is_valid"].tolist() == []
+        assert result["ret_code"].tolist() == []
+        assert result["comment"].tolist() == []
+        assert result["exception"].tolist() == []
+
 
 class TestValidationWorkflow:
     """Test the data_validation_framework.task.ValidationWorkflow class."""

tests/test_util.py

@@ -133,3 +133,13 @@ def test_apply_to_df(nb_processes, redirect_stdout):
         re.match(r"  File \"(\/.*?\.[\w:]+)\", line \d+, in _tested_func", exception_lines[3])
         is not None
     )
+
+    # Test with an empty DF
+    emty_res = util.apply_to_df(
+        df.loc[[]],
+        _tested_func,
+        "val1",
+        "val2",
+    )
+
+    pd.testing.assert_frame_equal(emty_res, df.loc[[]])