POST-REVIEW: logic around existing ingest versions

Author: fvankrieken

dcpy/connectors/ingest_datastore.py

@@ -74,7 +74,10 @@ def push_versioned(self, key: str, version: str, **kwargs) -> dict:
     def pull_versioned(
         self, key: str, version: str, destination_path: Path, **kwargs
     ) -> dict:
-        raise NotImplementedError
+        return self.storage.pull(
+            f"{key}/{version}/{key}.parquet",
+            destination_path / key / version,  # TODO a little hacky
+        )
 
     def list_versions(self, key: str, *, sort_desc: bool = True, **kwargs) -> list[str]:
         """This is maybe a problem in my plan"""
@@ -88,9 +91,17 @@ def _get_config_obj(self, key: str, version: str) -> dict:
             with open(Path(tmp_dir) / config_filename, "r", encoding="utf-8") as raw:
                 return yaml.safe_load(raw.read())
 
-    def get_config(self, key: str, version: str):
+    def get_config(self, key: str, version: str) -> ingest.Config:
         return ingest.Config(**self._get_config_obj(key, version))
 
+    def try_get_config(self, key: str, version: str) -> ingest.Config | None:
+        """for backwards compatibility"""
+        obj = self._get_config_obj(key, version)
+        if "dataset" not in obj:  # very specific to library
+            return ingest.Config(**obj)
+        else:
+            return None
+
     def get_latest_version(self, key: str, **kwargs) -> str:
         return self._get_config_obj(key, "latest")["version"]
 

dcpy/lifecycle/ingest/run.py

@@ -99,10 +99,12 @@ def ingest(
     shutil.copy(dataset_staging_dir / CONFIG_FILENAME, dataset_output_dir)
     shutil.copy(dataset_staging_dir / config.filename, dataset_output_dir)
 
-    is_new = validate.validate_against_existing_versions(
-        config.dataset, dataset_staging_dir / config.filename
-    )
-    if push and is_new:
+    if processed_datastore.version_exists(dataset_id, config.version):
+        validate.validate_against_existing_version(
+            dataset_id, config.version, dataset_staging_dir / config.filename
+        )
+        logger.info("Skipping archival")
+    elif push:
         assert config.archival.acl
         processed_datastore.push(
             dataset_id,
@@ -112,6 +114,4 @@ def ingest(
             overwrite=False,  ## TODO - allow this via flag?
             latest=latest,
         )
-    else:
-        logger.info("Skipping archival")
     return config

dcpy/lifecycle/ingest/validate.py

@@ -1,43 +1,39 @@
 import pandas as pd
 from pathlib import Path
+from tempfile import TemporaryDirectory
 
-from dcpy.models.lifecycle.ingest import Config
-from dcpy.connectors.edm import recipes
 from dcpy.utils.logging import logger
+from dcpy.lifecycle.ingest.connectors import processed_datastore
 
 
-def validate_against_existing_versions(ds: recipes.Dataset, filepath: Path) -> bool:
+def validate_against_existing_version(ds: str, version: str, filepath: Path) -> None:
     """
-    This function is called after a dataset has been preprocessed, just before archival
+    This function is called after a dataset has been processed, just before archival
     It's called in the case that the version of the dataset in the config (either provided or calculated)
-      already exists
+    already exists
 
     The last archived dataset with the same version is pulled in by pandas and compared to what was just processed
-    If they are identical, the last archived dataset has its config updated to reflect that it was checked but not re-archived
-    If they differ, the version is "patched" and a new patched version is archived
+    If they differ, an error is raised
     """
-    existing_config = recipes.try_get_config(ds)
-    if not existing_config:
-        logger.info(f"Dataset '{ds.key}' does not exist in recipes bucket")
-        return True
-    else:
-        if isinstance(existing_config, Config):
+    existing_config = processed_datastore.try_get_config(ds, version)
+    if existing_config:
+        with TemporaryDirectory() as tmp:
+            existing_file = processed_datastore.pull_versioned(ds, version, Path(tmp))[
+                "path"
+            ]
             new = pd.read_parquet(filepath)
-            comparison = recipes.read_df(ds)
-            if new.equals(comparison):
+            existing = pd.read_parquet(existing_file)
+            if new.equals(existing):
                 logger.info(
-                    f"Dataset '{ds.key}' already exists and matches newly processed data"
+                    f"Dataset id='{ds}' version='{version}' already exists and matches newly processed data"
                 )
-                return False
             else:
                 raise FileExistsError(
-                    f"Archived dataset '{ds.key}' already exists and has different data."
+                    f"Archived dataset id='{ds}' version='{version}' already exists and has different data."
                 )
 
-        # if previous was archived with library, we both expect some potential slight changes
-        # and are not able to update "freshness"
-        else:
-            logger.warning(
-                f"previous version of '{ds.key}' archived is from library. cannot update freshness"
-            )
-            return False
+    # if previous was archived with library, we both expect some potential slight changes and will not compare
+    else:
+        logger.warning(
+            f"Config of existing dataset id='{ds}' version='{version}' cannot be parsed."
+        )

dcpy/test/lifecycle/ingest/test_run.py

@@ -114,7 +114,7 @@ def test_run_repeat_version_fails_if_data_diff(
         patch_read_df.return_value = gpd.GeoDataFrame({"a": [None]}).set_geometry("a")
         with pytest.raises(
             FileExistsError,
-            match=f"Archived dataset 'id='{DATASET}' version='{FAKE_VERSION}'' already exists and has different data.",
+            match=f"Archived dataset id='{DATASET}' version='{FAKE_VERSION}' already exists and has different data.",
         ):
             run_ingest(
                 dataset_id=DATASET,

dcpy/test/lifecycle/ingest/test_validate.py

@@ -7,10 +7,10 @@
 from dcpy.models.lifecycle.ingest import Template
 from dcpy.utils import s3
 from dcpy.connectors.edm import recipes
+from dcpy.lifecycle.ingest.connectors import processed_datastore
 from dcpy.lifecycle.ingest import transform, validate
 
 from .shared import (
-    TEST_DATASET,
     TEST_OUTPUT,
     BASIC_CONFIG,
     BASIC_LIBRARY_CONFIG,
@@ -36,9 +36,6 @@ def test_validate_all_templates(dataset):
 
 
 class TestValidateAgainstExistingVersions:
-    def test_new(self, create_buckets):
-        assert validate.validate_against_existing_versions(TEST_DATASET, TEST_OUTPUT)
-
     def test_existing_library(self, create_buckets):
         ds = BASIC_LIBRARY_CONFIG.sparse_dataset
         config_str = json.dumps(BASIC_LIBRARY_CONFIG.model_dump(mode="json"))
@@ -48,20 +45,20 @@ def test_existing_library(self, create_buckets):
             f"{recipes.s3_folder_path(ds)}/config.json",
             BASIC_LIBRARY_CONFIG.dataset.acl,
         )
-        assert recipes.exists(ds)
-        assert not validate.validate_against_existing_versions(ds, TEST_OUTPUT)
+        assert processed_datastore.exists(ds.id, ds.version)
+        validate.validate_against_existing_version(ds.id, ds.version, TEST_OUTPUT)
 
     def test_existing(self, create_buckets):
         ds = BASIC_CONFIG.dataset
         recipes.archive_dataset(BASIC_CONFIG, TEST_OUTPUT, acl="private")
-        assert recipes.exists(ds)
-        assert not validate.validate_against_existing_versions(ds, TEST_OUTPUT)
+        assert processed_datastore.exists(ds.id, ds.version)
+        validate.validate_against_existing_version(ds.id, ds.version, TEST_OUTPUT)
 
     def test_existing_data_diffs(self, create_buckets):
         ds = BASIC_CONFIG.dataset
         recipes.archive_dataset(BASIC_CONFIG, TEST_OUTPUT, acl="private")
-        assert recipes.exists(ds)
+        assert processed_datastore.exists(ds.id, ds.version)
         with pytest.raises(FileExistsError):
-            validate.validate_against_existing_versions(
-                ds, TEST_OUTPUT.parent / "test.parquet"
+            validate.validate_against_existing_version(
+                ds.id, ds.version, TEST_OUTPUT.parent / "test.parquet"
             )