[FIX] logic for ingestion (#909)

* fix logic for ingestion

* change name

* make logic more robust

Author: jdkent

store/neurostore/ingest/extracted_features.py

@@ -10,10 +10,11 @@
     Pipeline,
     PipelineConfig,
     PipelineStudyResult,
+    BaseStudy,
 )
 
 
-def ingest_feature(feature_directory):
+def ingest_feature(feature_directory, overwrite=False):
     """Ingest demographics data into the database."""
     # read pipeline_info.json from the base feature directory
     with open(op.join(feature_directory, "pipeline_info.json")) as f:
@@ -23,21 +24,19 @@ def ingest_feature(feature_directory):
     pipeline = (
         db.session.query(Pipeline)
         .filter(
-            Pipeline.name == pipeline_info["name"],
+            Pipeline.name == pipeline_info["extractor"],
         )
         .first()
     )
     # create a pipeline if it does not exist
     if not pipeline:
         pipeline = Pipeline(
-            name=pipeline_info["name"],
+            name=pipeline_info["extractor"],
             description=pipeline_info.get("description"),
-            study_dependent=(
-                True if pipeline_info.get("type", False) == "dependent" else False
-            ),
-            ace_compatible="ace" in pipeline_info.get("input_sources", []),
-            pubget_compatible="pubget" in pipeline_info.get("input_sources", []),
-            derived_from=pipeline_info.get("derived_from", None),
+            study_dependent=None,
+            ace_compatible=True,
+            pubget_compatible=True,
+            derived_from=pipeline_info.get("input_pipelines", None),
         )
         db.session.add(pipeline)
 
@@ -56,10 +55,8 @@ def ingest_feature(feature_directory):
     if not pipeline_config:
         # Build config_args from pipeline_info
         config_args = {
-            "extractor": pipeline_info.get("extractor"),
             "extractor_kwargs": pipeline_info.get("extractor_kwargs", {}),
             "transform_kwargs": pipeline_info.get("transform_kwargs", {}),
-            "input_pipelines": pipeline_info.get("input_pipelines", {}),
         }
 
         pipeline_config = PipelineConfig(
@@ -76,27 +73,65 @@ def ingest_feature(feature_directory):
 
     # for each subject directory, read the results.json file and the info.json file
     pipeline_study_results = []
-    for paper_dir in paper_dirs:
-        with open(op.join(paper_dir, "results.json")) as f:
-            results = json.load(f)
-
-        with open(op.join(paper_dir, "info.json")) as f:
-            info = json.load(f)
 
+    for paper_dir in paper_dirs:
         # use the directory name as the base_study_id
         base_study_id = paper_dir.name
-        # create a new result record
-        pipeline_study_results.append(
-            PipelineStudyResult(
-                base_study_id=base_study_id,
-                result_data=results,
-                date_executed=parse_date(info["date"]),
-                file_inputs=info["inputs"],
-                config=pipeline_config,
-                status="SUCCESS",
+
+        if BaseStudy.query.filter_by(id=base_study_id).first() is None:
+            print(
+                f"Skipping {paper_dir} as it does not correspond to a valid base_study_id"
             )
+            continue
+        try:
+            with open(op.join(paper_dir, "results.json")) as f:
+                results = json.load(f)
+        except FileNotFoundError:
+            print(f"Skipping {paper_dir} as it does not contain results.json")
+            continue
+        except json.JSONDecodeError:
+            print(f"Skipping {paper_dir} as it contains invalid JSON in results.json")
+            continue
+        try:
+            with open(op.join(paper_dir, "info.json")) as f:
+                info = json.load(f)
+        except FileNotFoundError:
+            print(f"Skipping {paper_dir} as it does not contain info.json")
+            continue
+        except json.JSONDecodeError:
+            print(f"Skipping {paper_dir} as it contains invalid JSON in info.json")
+            continue
+
+        # check for existing result
+        existing_result = (
+            db.session.query(PipelineStudyResult)
+            .filter(
+                PipelineStudyResult.base_study_id == base_study_id,
+                PipelineStudyResult.config_id == pipeline_config.id,
+            )
+            .first()
         )
 
-    db.session.add_all(pipeline_study_results)
+        if existing_result and overwrite:
+            # update existing record
+            existing_result.result_data = results
+            existing_result.date_executed = parse_date(info["date"])
+            existing_result.file_inputs = info["inputs"]
+            existing_result.status = "SUCCESS"
+        elif not existing_result:
+            # create a new result record
+            pipeline_study_results.append(
+                PipelineStudyResult(
+                    base_study_id=base_study_id,
+                    result_data=results,
+                    date_executed=parse_date(info["date"]),
+                    file_inputs=info["inputs"],
+                    config=pipeline_config,
+                    status="SUCCESS",
+                )
+            )
+
+    if pipeline_study_results:
+        db.session.add_all(pipeline_study_results)
 
     db.session.commit()