Merge pull request #482 from pepkit/dev_destroy_pipestat_results

Dev destroy pipestat results

looper/looper.py

@@ -46,7 +46,6 @@
     sample_folder,
 )
 from pipestat.reports import get_file_for_table
-from pipestat.reports import get_file_for_project
 
 _PKGNAME = "looper"
 _LOGGER = logging.getLogger(_PKGNAME)
@@ -242,6 +241,20 @@ def __call__(self, args, preview_flag=True):
         :param bool preview_flag: whether to halt before actually removing files
         """
 
+        use_pipestat = (
+            self.prj.pipestat_configured_project
+            if getattr(args, "project", None)
+            else self.prj.pipestat_configured
+        )
+
+        if use_pipestat:
+            _LOGGER.info("Removing summary:")
+            destroy_summary(
+                self.prj,
+                dry_run=preview_flag,
+                project_level=getattr(args, "project", None),
+            )
+
         _LOGGER.info("Removing results:")
 
         for sample in select_samples(prj=self.prj, args=args):
@@ -251,26 +264,16 @@ def __call__(self, args, preview_flag=True):
                 # Preview: Don't actually delete, just show files.
                 _LOGGER.info(str(sample_output_folder))
             else:
-                _remove_or_dry_run(sample_output_folder, args.dry_run)
-
-        _LOGGER.info("Removing summary:")
-        use_pipestat = (
-            self.prj.pipestat_configured_project
-            if getattr(
-                args, "project", None
-            )  # TODO this argument hasn't been added to the pydantic models.
-            else self.prj.pipestat_configured
-        )
-        if use_pipestat:
-            destroy_summary(
-                self.prj,
-                getattr(args, "dry_run", None),
-                getattr(args, "project", None),
-            )
-        else:
-            _LOGGER.warning(
-                "Pipestat must be configured to destroy any created summaries."
-            )
+                if use_pipestat:
+                    psms = self.prj.get_pipestat_managers(
+                        sample_name=sample.sample_name
+                    )
+                    for pipeline_name, psm in psms.items():
+                        psm.backend.remove_record(
+                            record_identifier=sample.sample_name, rm_record=True
+                        )
+                else:
+                    _remove_or_dry_run(sample_output_folder, args.dry_run)
 
         if not preview_flag:
             _LOGGER.info("Destroy complete.")
@@ -692,10 +695,8 @@ def destroy_summary(prj, dry_run=False, project_level=False):
         for name, psm in psms.items():
             _remove_or_dry_run(
                 [
-                    get_file_for_project(
-                        psm,
-                        pipeline_name=psm.pipeline_name,
-                        directory="reports",
+                    get_file_for_table(
+                        psm, pipeline_name=psm.pipeline_name, directory="reports"
                     ),
                     get_file_for_table(
                         psm,
@@ -707,8 +708,8 @@ def destroy_summary(prj, dry_run=False, project_level=False):
                         pipeline_name=psm.pipeline_name,
                         appendix="objs_summary.yaml",
                     ),
-                    get_file_for_table(
-                        psm, pipeline_name=psm.pipeline_name, appendix="reports"
+                    os.path.join(
+                        os.path.dirname(psm.config_path), "aggregate_results.yaml"
                     ),
                 ],
                 dry_run,
@@ -724,10 +725,8 @@ def destroy_summary(prj, dry_run=False, project_level=False):
             for name, psm in psms.items():
                 _remove_or_dry_run(
                     [
-                        get_file_for_project(
-                            psm,
-                            pipeline_name=psm.pipeline_name,
-                            directory="reports",
+                        get_file_for_table(
+                            psm, pipeline_name=psm.pipeline_name, directory="reports"
                         ),
                         get_file_for_table(
                             psm,
@@ -739,8 +738,8 @@ def destroy_summary(prj, dry_run=False, project_level=False):
                             pipeline_name=psm.pipeline_name,
                             appendix="objs_summary.yaml",
                         ),
-                        get_file_for_table(
-                            psm, pipeline_name=psm.pipeline_name, appendix="reports"
+                        os.path.join(
+                            os.path.dirname(psm.config_path), "aggregate_results.yaml"
                         ),
                     ],
                     dry_run,

requirements/requirements-all.txt

@@ -6,7 +6,7 @@ logmuse>=0.2.0
 pandas>=2.0.2
 pephubclient>=0.4.0
 peppy>=0.40.0
-pipestat>=0.8.2a1
+pipestat>=0.8.3a1
 pyyaml>=3.12
 rich>=9.10.0
 ubiquerg>=0.5.2

tests/test_comprehensive.py

@@ -1,3 +1,5 @@
+import os.path
+
 import pytest
 from peppy.const import *
 from yaml import dump
@@ -10,6 +12,7 @@
 from tests.smoketests.test_run import is_connected
 from tempfile import TemporaryDirectory
 from pipestat import PipestatManager
+from pipestat.exceptions import RecordNotFoundError
 
 from yaml import dump, safe_load
 
@@ -137,3 +140,9 @@ def test_comprehensive_looper_pipestat(prep_temp_pep_pipestat):
         result = main(test_args=x)
     except Exception:
         raise pytest.fail("DID RAISE {0}".format(Exception))
+
+    sd = os.path.dirname(path_to_looper_config)
+    tsv_list = [os.path.join(sd, f) for f in os.listdir(sd) if f.endswith(".tsv")]
+    assert len(tsv_list) == 0
+    with pytest.raises(RecordNotFoundError):
+        retrieved_result = psm.retrieve_one(record_identifier="frog_2")