Refactor copy paths

Author: steve-sullivan

mokelumne/dags/copy_files.py

@@ -15,7 +15,10 @@
 
 from mokelumne.util.storage import run_dir
 from mokelumne.util.file_transfer import (
+    DESTINATION_VOLUMES,
+    SOURCE_VOLUMES,
     build_file_manifest,
+    build_volume_path,
     clean_destination_path,
     copy_files_from_manifest,
     load_json,
@@ -25,40 +28,6 @@
 
 logger = logging.getLogger(__name__)
 
-SOURCE_VOLUMES = [
-    "/srv/da",
-    "/srv/pa",
-    "/srv/lpsdata4",
-]
-
-DESTINATION_VOLUMES = [
-    "/srv/pa",
-    "/srv/da",
-]
-
-
-def build_volume_path(
-    volume: str,
-    subdirectory: str,
-    label: str,
-) -> Path:
-    """Build a path from a volume and relative subdirectory."""
-
-    if subdirectory == "":
-        raise AirflowFailException(
-            f"{label} subdirectory is required"
-        )
-
-    subdirectory_path = Path(subdirectory)
-
-    if subdirectory_path.is_absolute():
-        raise AirflowFailException(
-            f"{label} subdirectory must be relative: {subdirectory}"
-        )
-
-    return Path(volume) / subdirectory_path
-
-
 @dag(
     description="Transfers files from one location to another",
     schedule=None,
@@ -88,7 +57,7 @@ def build_volume_path(
             default="",
             type="string",
             title="Destination subdirectory",
-            description="Relative path within the destination volume. Do not include leading '/'.",
+            description="Relative path within the destination volume. Do not include leading '/' or the final '/incoming' to the path.",
         ),
         "exclude_regex": Param(
             default="",
@@ -105,18 +74,39 @@ def copy_files():
     """Copy files from a source directory to an empty destination directory."""
 
     @task
-    def validate_source() -> str:
-        """Checks that the source is a valid directory."""
+    def build_copy_paths() -> dict[str, str]:
+        """Build the resolved source and destination paths."""
 
         ctx = get_current_context()
-        source_volume = ctx["params"]["source_volume"]
-        source_subdirectory = ctx["params"]["source_subdirectory"]
 
-        source_path = build_volume_path(
-            source_volume,
-            source_subdirectory,
-            "Source",
-        )
+        try:
+            source_path = build_volume_path(
+                ctx["params"]["source_volume"],
+                ctx["params"]["source_subdirectory"],
+                "Source",
+            )
+
+            destination_path = build_volume_path(
+                ctx["params"]["destination_volume"],
+                ctx["params"]["destination_subdirectory"],
+                "Destination",
+            )
+
+            destination_path = clean_destination_path(destination_path)
+        except ValueError as ex:
+            raise AirflowFailException(str(ex)) from ex
+
+        return {
+            "source": str(source_path),
+            "destination": str(destination_path)
+        }
+
+
+    @task
+    def validate_source(copy_paths: dict[str, str]) -> str:
+        """Checks that the source is a valid directory."""
+
+        source_path = Path(copy_paths["source"])
 
         if not source_path.exists():
             raise AirflowFailException(f"Source path does not exist: {source_path}")
@@ -127,19 +117,10 @@ def validate_source() -> str:
         return str(source_path)
 
     @task
-    def prepare_destination() -> str:
+    def prepare_destination(copy_paths: dict[str, str]) -> str:
         """Prepare the destination directory."""
 
-        ctx = get_current_context()
-        destination_volume = ctx["params"]["destination_volume"]
-        destination_subdirectory = ctx["params"]["destination_subdirectory"]
-
-        destination_path = build_volume_path(
-            destination_volume,
-            destination_subdirectory,
-            "Destination",
-        )
-        destination_path = clean_destination_path(destination_path)
+        destination_path = Path(copy_paths["destination"])
 
         if not destination_path.exists():
             logger.info("Creating destination directory: %s", destination_path)
@@ -153,6 +134,7 @@ def prepare_destination() -> str:
 
         return str(destination_path)
 
+
     @task
     def build_manifest(source: str) -> str:
         """Build a manifest of all files under the source directory."""
@@ -218,19 +200,24 @@ def verify_manifest(destination: str, manifest_path: str) -> None:
         logger.info("Verification report written to: %s", verification_report_path)
         logger.info("Verified %s copied file(s)", len(verification_report))
 
-    """The user needs to confirm the file copy paths before proceeding"""
+
+    # Need to run this before defining confirm_copy since we need the resolved paths:
+    copy_paths = build_copy_paths()
+
+    # The user needs to confirm the file copy paths before proceeding
     confirm_copy = ApprovalOperator(
         task_id="confirm_copy",
         subject="Review the copy operation and approve it to continue.",
         body=(
             "Approve file copy from "
-            "**{{ params.source_volume }}/{{ params.source_subdirectory }}** "
+            f"**{copy_paths['source']}** "
             "to "
-            "**{{ params.destination_volume }}/{{ params.destination_subdirectory }}**"
+            f"**{copy_paths['destination']}**"
         ),
     )
-    validated_source = validate_source()
-    prepared_destination = prepare_destination()
+
+    validated_source = validate_source(copy_paths)
+    prepared_destination = prepare_destination(copy_paths)
     manifest = build_manifest(validated_source)
     copied_manifest_files = copy_manifest_files(
         validated_source,
@@ -240,7 +227,8 @@ def verify_manifest(destination: str, manifest_path: str) -> None:
     verified_manifest = verify_manifest(prepared_destination, manifest)
 
     (
-        validated_source
+        copy_paths
+        >> validated_source
         >> prepared_destination
         >> manifest
         >> confirm_copy

mokelumne/util/file_transfer.py

@@ -10,6 +10,37 @@
 ManifestEntry = dict[str, int | str]
 Manifest = dict[str, str | list[ManifestEntry]]
 
+SOURCE_VOLUMES = [
+    "/srv/lpsdata4",
+    "/srv/dcushare",
+    "/srv/lpsdata2",
+    "/srv/ealdata",
+    "/srv/rohoshare",
+]
+
+DESTINATION_VOLUMES = [
+    "/srv/da",
+    "/srv/pa",
+]
+
+
+def build_volume_path(
+    volume: str,
+    subdirectory: str,
+    label: str,
+) -> Path:
+    """Build a path from a volume and relative subdirectory."""
+
+    if subdirectory == "":
+        raise ValueError(f"{label} subdirectory is required")
+
+    subdirectory_path = Path(subdirectory)
+
+    if subdirectory_path.is_absolute():
+        raise ValueError(f"{label} subdirectory must be relative: {subdirectory}")
+
+    return Path(volume) / subdirectory_path
+
 
 def sha256_for_file(path: Path) -> str:
     """Calculate the SHA256 checksum for a file."""
@@ -61,10 +92,7 @@ def clean_destination_path(destination_path: Path) -> Path:
     return Path(destination_path / "incoming")
 
 
-def verify_file_manifest(
-        destination_path: Path,
-        manifest_path: Path
-) -> list[ManifestEntry]:
+def verify_file_manifest(destination_path: Path, manifest_path: Path) -> list[ManifestEntry]:
     """Verify all copied files exist at the destination."""
 
     verification_report: list[ManifestEntry] = []
@@ -111,11 +139,7 @@ def verify_file_manifest(
     return verification_report
 
 
-def copy_files_from_manifest(
-    source_path: Path,
-    destination_path: Path,
-    manifest_path: Path,
-) -> None:
+def copy_files_from_manifest(source_path: Path, destination_path: Path, manifest_path: Path) -> None:
     """Copy all files in manifest to destination directory."""
 
     manifest = load_json(manifest_path)

test/unit/test_file_transfer.py

@@ -60,6 +60,37 @@ def test_save_and_load_json(self, tmp_path: Path):
 
         assert result == data
 
+    def test_build_volume_path(self):
+        """Ensure build_volume_path joins a volume and relative subdirectory."""
+
+        result = file_transfer.build_volume_path(
+            "tmp/srv/pa",
+            "aerial/ucb",
+            "Source",
+        )
+
+        assert result == Path("tmp/srv/pa/aerial/ucb")
+
+    def test_build_volume_path_requires_subdirectory(self):
+        """Ensure build_volume_path requires a subdirectory."""
+
+        with pytest.raises(ValueError, match="Source subdirectory is required"):
+            file_transfer.build_volume_path(
+                "tmp/srv/pa",
+                "",
+                "Source",
+            )
+
+    def test_build_volume_path_rejects_absolute_subdirectory(self):
+        """Ensure build_volume_path rejects absolute subdirectories."""
+
+        with pytest.raises(ValueError, match="Source subdirectory must be relative"):
+            file_transfer.build_volume_path(
+                "tmp/srv/pa",
+                "/aerial/ucb",
+                "Source",
+            )
+
     def test_build_manifest_empty_directory(self, tmp_path: Path):
         """Ensure that build_manifest returns an empty list for an empty directory."""
         result = file_transfer.build_file_manifest(tmp_path)