Introduce transfer modes and implement move+symlink (#1384)

* enforce read_only on remote datasets

* reset state to server state when change was tried.

* add tests for changing properties on remote datasets. fix test.py

* format

* introduce transfer modes.

* introduce transfer mode. add test for move+symlink

* move transfer mode to own file

* remove no longer used file

* improve comment

* fix test

* add changelog

---------

Co-authored-by: valentin-pinkau <[email protected]>

Author: valentin-pinkau

webknossos/Changelog.md

@@ -41,6 +41,7 @@ For upgrade instructions, please check the respective _Breaking Changes_ section
 - Updated the api version of the webknossos-api to 12. [#1371](https://github.com/scalableminds/webknossos-libs/pull/1371])
 - Allowing RemoteDataset to align mags, when down- or upsampling [#1382](https://github.com/scalableminds/webknossos-libs/pull/1382)
 - RemoteDatasets that use zarr streaming are no longer read-only. [#1383](https://github.com/scalableminds/webknossos-libs/pull/1383)
+- Introduced transfer modes COPY, MOVE_AND_SYMLINK and HTTP for Dataset.upload() [#1384](https://github.com/scalableminds/webknossos-libs/pull/1384)
 
 ### Fixed
 - Fixed test.py to parse the command line arguments correctly. [#1383](https://github.com/scalableminds/webknossos-libs/pull/1383)

webknossos/tests/dataset/test_dataset_download_upload_remote.py

@@ -6,6 +6,7 @@
 from upath import UPath
 
 import webknossos as wk
+from webknossos import TransferMode
 
 SAMPLE_BBOX = wk.BoundingBox((3164, 3212, 1017), (10, 10, 10))
 
@@ -100,8 +101,7 @@ def test_upload_dataset_with_symlinks(tmp_upath: UPath) -> None:
     sample_dataset = get_sample_dataset(tmp_upath)
     remote_ds = sample_dataset.upload(
         new_dataset_name="test_remote_symlink",
-        upload_directly_to_common_storage=True,
-        symlink_data_instead_of_copy=True,
+        transfer_mode=TransferMode.MOVE_AND_SYMLINK,
     )
     assert np.array_equal(
         remote_ds.get_color_layers()[0].get_finest_mag().read(),
@@ -115,7 +115,7 @@ def test_upload_dataset_with_symlinks(tmp_upath: UPath) -> None:
 def test_upload_dataset_copy_to_paths(tmp_upath: UPath) -> None:
     sample_dataset = get_sample_dataset(tmp_upath)
     remote_ds = sample_dataset.upload(
-        new_dataset_name="test_remote_copy", upload_directly_to_common_storage=True
+        new_dataset_name="test_remote_copy", transfer_mode=TransferMode.COPY
     )
     assert np.array_equal(
         remote_ds.get_color_layers()[0].get_finest_mag().read(),

webknossos/tests/dataset/test_move_symlink.py

@@ -0,0 +1,38 @@
+import tempfile
+
+import pytest
+from upath import UPath
+
+from webknossos import TransferMode
+
+
+def test_transfer_mode_move_and_symlink() -> None:
+    transfer_mode = TransferMode.MOVE_AND_SYMLINK
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        src = UPath(tmp_dir) / "src_test"
+        dst = UPath(tmp_dir) / "dst_test"
+
+        src.mkdir(parents=True, exist_ok=True)
+        (src / "a.txt").touch()
+
+        transfer_mode.transfer(src, dst)
+
+        assert (dst / "a.txt").exists()
+        assert (src / "a.txt").exists()
+        assert src.is_symlink()
+
+        # try to move again to same destination
+        with pytest.raises(AssertionError):
+            transfer_mode.transfer(src, dst)
+
+        # try to move again to different destination
+        dst2 = UPath(tmp_dir) / "dst_test2"
+        transfer_mode.transfer(src, dst2)
+
+        assert (dst / "a.txt").exists()
+        assert not dst.is_symlink()
+        assert (dst2 / "a.txt").exists()
+        assert dst2.is_symlink()
+        assert (src / "a.txt").exists()
+        assert src.is_symlink()

webknossos/webknossos/dataset/__init__.py

@@ -25,3 +25,4 @@
 from .remote_dataset import RemoteDataset
 from .remote_folder import RemoteFolder
 from .sampling_modes import SamplingModes
+from .transfer_mode import TransferMode

webknossos/webknossos/dataset/dataset.py

@@ -59,6 +59,7 @@
 from .remote_dataset import RemoteDataset
 from .remote_folder import RemoteFolder
 from .sampling_modes import SamplingModes
+from .transfer_mode import TransferMode
 
 if TYPE_CHECKING:
     import pims
@@ -565,10 +566,10 @@ def publish_to_preliminary_dataset(
         self,
         dataset_id: str,
         path_prefix: str | None = None,
-        symlink_data_instead_of_copy: bool = False,
+        transfer_mode: TransferMode = TransferMode.COPY,
     ) -> None:
         """
-        Copies or symlinks the data to paths returned by WEBKNOSSOS
+        Copies or moves+symlinks the data to paths returned by WEBKNOSSOS
         The dataset needs to be in status "uploading".
         The dataset already exists in WEBKNOSSOS but has no dataset_properties.
         With the dataset_properties WEBKNOSSOS can reserve the paths.
@@ -579,16 +580,17 @@ def publish_to_preliminary_dataset(
         """
         from ..client.context import _get_context
 
+        if transfer_mode == TransferMode.HTTP:
+            raise ValueError("HTTP transfer mode is not supported for this method")
+
         context = _get_context()
         response = context.api_client_with_auth.reserve_dataset_upload_to_paths_for_preliminary(
             dataset_id,
             ApiReserveDatasetUploadToPathsForPreliminaryParameters(
                 self._properties, path_prefix
             ),
         )
-        self._copy_or_symlink_dataset_to_paths(
-            response.data_source, symlink_data_instead_of_copy
-        )
+        self._copy_or_symlink_dataset_to_paths(response.data_source, transfer_mode)
 
         context.api_client_with_auth.finish_dataset_upload_to_paths(dataset_id)
 
@@ -599,10 +601,9 @@ def upload(
         folder_id: str | RemoteFolder | None = None,
         require_unique_name: bool = False,
         layers_to_link: Sequence[LayerToLink | RemoteLayer] | None = None,
-        upload_directly_to_common_storage: bool = False,
+        transfer_mode: TransferMode = TransferMode.HTTP,
         jobs: int | None = None,
         common_storage_path_prefix: str | None = None,
-        symlink_data_instead_of_copy: bool = False,
     ) -> "RemoteDataset":
         """Upload this dataset to webknossos.
 
@@ -615,11 +616,9 @@ def upload(
             folder_id: Optional ID of folder where dataset should be placed
             require_unique_name: Whether to make request fail in case a dataset with the name already exists
             layers_to_link: Optional list of LayerToLink to link already published layers to the dataset.
-            upload_directly_to_common_storage: Set this to true when the client has access to the same storage system as the WEBKNOSSOS datastore (file system or cloud storage).
             jobs: Optional number of jobs to use for uploading the data.
-            common_storage_path_prefix: Optional path prefix used when upload_directly_to_common_storage is true to select one of the available mount points for the dataset folder.
-            symlink_data_instead_of_copy: Considered, when upload_directly_to_common_storage is True. Set this to true when the client has access to the same file system as the WEBKNOSSOS datastore.
-            When set to true, the data symlinked to the new location.
+            common_storage_path_prefix: Optional path prefix used when transfer_mode is either COPY or MOVE_AND_SYMLINK
+                                        to select one of the available WEBKNOSSOS storages.
         Returns:
             RemoteDataset: Reference to the newly created remote dataset
         Note:
@@ -641,11 +640,6 @@ def upload(
             ```
         """
 
-        if symlink_data_instead_of_copy:
-            assert upload_directly_to_common_storage, (
-                "Cannot use symlinking with upload_directly_to_common_storage=False"
-            )
-
         from ..client.context import _get_context
 
         new_dataset_name = new_dataset_name or self.name
@@ -662,7 +656,7 @@ def upload(
             ]
         )
 
-        if upload_directly_to_common_storage:
+        if transfer_mode in (TransferMode.COPY, TransferMode.MOVE_AND_SYMLINK):
             context = _get_context()
             response = context.api_client_with_auth.reserve_dataset_upload_to_paths(
                 ApiReserveDatasetUplaodToPathsParameters(
@@ -683,12 +677,16 @@ def upload(
             new_dataset_id = response.new_dataset_id
             data_source = response.data_source
 
-            self._copy_or_symlink_dataset_to_paths(
-                data_source, symlink_data_instead_of_copy
-            )
+            self._copy_or_symlink_dataset_to_paths(data_source, transfer_mode)
             # announce finished upload
             context.api_client_with_auth.finish_dataset_upload_to_paths(new_dataset_id)
         else:
+            assert transfer_mode == TransferMode.HTTP, "Expected HTTP transfer mode"
+            if common_storage_path_prefix is not None:
+                logger.warning(
+                    "common_storage_path_prefix is not None, but uploading via HTTP."
+                    " Ignoring common_storage_path_prefix."
+                )
             for layer in self.get_segmentation_layers():
                 if not layer.attachments.is_empty:
                     raise NotImplementedError(
@@ -704,31 +702,23 @@ def upload(
         return RemoteDataset.open(dataset_id=new_dataset_id)
 
     def _copy_or_symlink_dataset_to_paths(
-        self, data_source: DatasetProperties, symlink_data_instead_of_copy: bool
+        self, data_source: DatasetProperties, transfer_mode: TransferMode
     ) -> None:
         """
-        Iterates over the mags and attachments and copies or symlinks them to the target location.
+        Iterates over the mags and attachments and copies or moves then symlinks them to the target location.
         """
-        # copy data
+        assert transfer_mode in (TransferMode.COPY, TransferMode.MOVE_AND_SYMLINK), (
+            f"transfer mode not supported. found {transfer_mode}"
+        )
+
+        # transfer data
         for layer in data_source.data_layers:
             src_layer = self.layers[layer.name]
             for mag in layer.mags:
                 src_mag = src_layer.mags[mag.mag]
                 assert mag.path is not None, "mag.path must be set to copy/move data"
                 dst_mag_path = enrich_path(mag.path)
-                if symlink_data_instead_of_copy:
-                    assert is_fs_path(src_mag.path) and is_fs_path(dst_mag_path), (
-                        f"Either src_mag.path or mag.path are not pointing to a local file system {src_mag.path}, {mag.path}"
-                    )
-                    dst_mag_path.parent.mkdir(parents=True, exist_ok=True)
-                    dst_mag_path.unlink(missing_ok=True)
-                    dst_mag_path.symlink_to(src_mag.path)
-                else:
-                    copytree(
-                        src_mag.path,
-                        dst_mag_path,
-                        progress_desc=f"copying mag {src_mag.path} to {mag.path}",
-                    )
+                transfer_mode.transfer(src_mag.path, dst_mag_path)
             if isinstance(src_layer, SegmentationLayer):
                 assert isinstance(layer, SegmentationLayerProperties), (
                     "If src_layer is a SegmentationLayer, then layer must be a SegmentationLayerProperties"
@@ -738,20 +728,7 @@ def _copy_or_symlink_dataset_to_paths(
                     src_layer.attachments, layer.attachments
                 ):
                     dst_attachment_path = enrich_path(dst_attachment.path)
-                    if symlink_data_instead_of_copy:
-                        assert is_fs_path(src_attachment.path) and is_fs_path(
-                            dst_attachment_path
-                        ), (
-                            f"Either src_attachment.path or dst_attachment.path are not pointing to a local file system {src_attachment.path}, {dst_attachment.path}"
-                        )
-                        dst_attachment_path.unlink()
-                        dst_attachment_path.symlink_to(src_attachment.path)
-                    else:
-                        copytree(
-                            src_attachment.path,
-                            enrich_path(dst_attachment.path),
-                            progress_desc=f"copying attachment {src_attachment.path} to {dst_attachment.path}",
-                        )
+                    transfer_mode.transfer(src_attachment.path, dst_attachment_path)
 
     @staticmethod
     def get_remote_datasets(