move scp to scp_util.py

src/depiction_targeted_preprocbatch/job_export_results.py

@@ -1,15 +1,14 @@
 from __future__ import annotations
 
-import subprocess
 import zipfile
 from pathlib import Path
 
 from bfabric import Bfabric
 from bfabric.entities import Storage
-from loguru import logger
 
 from depiction.persistence.file_checksums import FileChecksums
 from depiction_targeted_preproc.app.workunit_config import WorkunitConfig
+from depiction_targeted_preprocbatch.scp_util import scp
 
 
 class JobExportResults:
@@ -53,13 +52,5 @@ def _copy_zip_to_storage(self, zip_file_path, output_storage):
         output_path = self._workunit_config.output_folder_absolute_path / zip_file_path.name
         output_path_relative = output_path.relative_to(output_storage.base_path)
         output_uri = f"{output_storage.scp_prefix}{output_path_relative}"
-        self._scp(zip_file_path, output_uri)
+        scp(zip_file_path, output_uri)
         return output_path_relative
-
-    def _scp(self, source: str | Path, target: str | Path) -> None:
-        """Performs scp source target.
-        Make sure that either the source or target specifies a host, otherwise you should just use shutil.copyfile.
-        """
-        # TODO this should be moved to a central location
-        logger.info(f"scp {source} {target}")
-        subprocess.run(["scp", source, target], check=True)

src/depiction_targeted_preprocbatch/job_prepare_inputs.py

@@ -1,14 +1,12 @@
 from __future__ import annotations
 
 import shutil
-import subprocess
 from pathlib import Path
 from typing import TYPE_CHECKING
 
-from loguru import logger
-
 from depiction.persistence.file_checksums import FileChecksums
 from depiction_targeted_preproc.pipeline.setup import write_standardized_table
+from depiction_targeted_preprocbatch.scp_util import scp
 
 if TYPE_CHECKING:
     from depiction_targeted_preprocbatch.executor import BatchJob
@@ -56,7 +54,7 @@ def stage_imzml(self) -> None:
 
         # perform the copies
         for scp_uri, result_name in zip(scp_uris, ["raw.imzML", "raw.ibd"]):
-            self._scp(scp_uri, str(self._sample_dir / result_name))
+            scp(scp_uri, str(self._sample_dir / result_name))
 
         # check the checksum
         actual_checksum = FileChecksums(file_path=self._sample_dir / "raw.imzML").checksum_md5
@@ -73,11 +71,3 @@ def stage_pipeline_parameters(self) -> None:
             self._sample_dir.parents[1] / "pipeline_params.yml",
             self._sample_dir / "pipeline_params.yml",
         )
-
-    def _scp(self, source: str | Path, target: str | Path) -> None:
-        """Performs scp source target.
-        Make sure that either the source or target specifies a host, otherwise you should just use shutil.copyfile.
-        """
-        # TODO this should be moved to a central location
-        logger.info(f"scp {source} {target}")
-        subprocess.run(["scp", source, target], check=True)

src/depiction_targeted_preprocbatch/scp_util.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+
+from loguru import logger
+
+
+def scp(source: str | Path, target: str | Path) -> None:
+    """Performs scp source target.
+    Make sure that either the source or target specifies a host, otherwise you should just use shutil.copyfile.
+    """
+    logger.info(f"scp {source} {target}")
+    subprocess.run(["scp", source, target], check=True)