Merge branch 'nancodes' into nans_cdc_covidnet

Author: dshemetov

_delphi_utils_python/delphi_utils/archive.py

@@ -254,6 +254,15 @@ def diff_exports(self) -> Tuple[Files, FileDiffMap, Files]:
                 new_issues_df.to_csv(diff_file, na_rep="NA")
                 common_diffs[after_file] = diff_file
 
+        # Replace deleted files with empty versions, but only if the cached version is not
+        # already empty
+        for deleted_file in deleted_files:
+            deleted_df = pd.read_csv(deleted_file)
+            if not deleted_df.empty:
+                empty_df = deleted_df[0:0]
+                new_deleted_filename = join(self.export_dir, basename(deleted_file))
+                empty_df.to_csv(new_deleted_filename, index=False)
+
         return deleted_files, common_diffs, new_files
 
     def archive_exports(self, exported_files: Files) -> Tuple[Files, Files]:
@@ -280,9 +289,10 @@ def filter_exports(self, common_diffs: FileDiffMap):
         Filter export directory to only contain relevant files.
 
         Filters down the export_dir to only contain:
-        1) New files, 2) Changed files, filtered-down to the ADDED and CHANGED rows only.
-        Should be called after archive_exports() so we archive the raw exports before
-        potentially modifying them.
+        1) New files, 2) Changed files, filtered-down to the ADDED and CHANGED rows
+        only, and 3) Deleted files replaced with empty CSVs with the same name. Should
+        be called after archive_exports() so we archive the raw exports before potentially
+        modifying them.
 
         Parameters
         ----------
@@ -311,9 +321,9 @@ def run(self):
         self.update_cache()
 
         # Diff exports, and make incremental versions
-        _, common_diffs, new_files = self.diff_exports()
+        deleted_files, common_diffs, new_files = self.diff_exports()
 
-        # Archive changed and new files only
+        # Archive changed, new, and emptied deleted files
         to_archive = [f for f, diff in common_diffs.items()
                       if diff is not None]
         to_archive += new_files

_delphi_utils_python/tests/test_archive.py

@@ -134,6 +134,19 @@ def test_diff_and_filter_exports(self, tmp_path):
             "missing_sample_size": [Nans.DELETED] + [Nans.NOT_MISSING] * 2,
             })
 
+        csv2_deleted = pd.DataFrame(
+            np.empty(0, dtype=[
+                ("geo_id", str),
+                ("val", float),
+                ("se", float),
+                ("sample_size", float),
+                ("missing_val", int),
+                ("missing_se", int),
+                ("missing_sample_size", int)
+                ]),
+            index=[]
+            )
+
         arch_diff = ArchiveDiffer(cache_dir, export_dir)
 
         # Test diff_exports
@@ -163,7 +176,10 @@ def test_diff_and_filter_exports(self, tmp_path):
 
         # Check filesystem for actual files
         assert set(listdir(export_dir)) == {
-            "csv0.csv", "csv1.csv", "csv1.csv.diff", "csv3.csv", "csv4.csv", "csv4.csv.diff"}
+            "csv0.csv", "csv1.csv", "csv1.csv.diff",
+            "csv3.csv", "csv4.csv", "csv4.csv.diff",
+            "csv2.csv"
+        }
         assert_frame_equal(
             pd.read_csv(join(export_dir, "csv1.csv.diff"), dtype=CSV_DTYPES),
             csv1_diff)
@@ -180,8 +196,11 @@ def test_diff_and_filter_exports(self, tmp_path):
 
         arch_diff.filter_exports(common_diffs)
 
-        # Check exports directory just has incremental changes
-        assert set(listdir(export_dir)) == {"csv1.csv", "csv3.csv", "csv4.csv"}
+        # Check exports directory just has incremental and deleted changes
+        assert set(listdir(export_dir)) == {"csv1.csv", "csv2.csv", "csv3.csv", "csv4.csv"}
+        assert_frame_equal(
+            pd.read_csv(join(export_dir, "csv2.csv"), dtype=CSV_DTYPES),
+            csv2_deleted)
         assert_frame_equal(
             pd.read_csv(join(export_dir, "csv1.csv"), dtype=CSV_DTYPES),
             csv1_diff)
@@ -308,7 +327,7 @@ def test_run(self, tmp_path, s3_client):
             assert_frame_equal(pd.read_csv(body, dtype=CSV_DTYPES), df)
 
         # Check exports directory just has incremental changes
-        assert set(listdir(export_dir)) == {"csv1.csv", "csv3.csv", "csv4.csv"}
+        assert set(listdir(export_dir)) == {"csv1.csv", "csv2.csv", "csv3.csv", "csv4.csv"}
         csv1_diff = pd.DataFrame({
             "geo_id": ["3", "2", "4"],
             "val": [np.nan, 2.1, 4.0],
@@ -321,6 +340,21 @@ def test_run(self, tmp_path, s3_client):
         assert_frame_equal(
             pd.read_csv(join(export_dir, "csv1.csv"), dtype=CSV_DTYPES),
             csv1_diff)
+        csv2_deleted = pd.DataFrame(
+            np.empty(0, dtype=[
+                ("geo_id", str),
+                ("val", float),
+                ("se", float),
+                ("sample_size", float),
+                ("missing_val", int),
+                ("missing_se", int),
+                ("missing_sample_size", int)
+                ]),
+            index=[]
+            )
+        assert_frame_equal(
+            pd.read_csv(join(export_dir, "csv2.csv"), dtype=CSV_DTYPES),
+            csv2_deleted)
 
 
 class TestGitArchiveDiffer:
@@ -521,7 +555,7 @@ def test_run(self, tmp_path):
         original_branch.checkout()
 
         # Check exports directory just has incremental changes
-        assert set(listdir(export_dir)) == {"csv1.csv", "csv3.csv", "csv4.csv"}
+        assert set(listdir(export_dir)) == {"csv1.csv", "csv2.csv", "csv3.csv", "csv4.csv"}
         csv1_diff = pd.DataFrame({
             "geo_id": ["3", "2", "4"],
             "val": [np.nan, 2.1, 4.0],
@@ -534,6 +568,22 @@ def test_run(self, tmp_path):
         assert_frame_equal(
             pd.read_csv(join(export_dir, "csv1.csv"), dtype=CSV_DTYPES),
             csv1_diff)
+        csv2_deleted = pd.DataFrame(
+            np.empty(0, dtype=[
+                ("geo_id", str),
+                ("val", float),
+                ("se", float),
+                ("sample_size", float),
+                ("missing_val", int),
+                ("missing_se", int),
+                ("missing_sample_size", int)
+                ]),
+            index=[]
+            )
+        assert_frame_equal(
+            pd.read_csv(join(export_dir, "csv2.csv"), dtype=CSV_DTYPES),
+            csv2_deleted)
+
 
 
 class TestFromParams: