export_workspace: support not merging derived_from links

CHANGELOG.md

@@ -36,7 +36,7 @@ and start a new "In Progress" section above it.
 - Use `save_result` node id in synchronous download filename hint (content-disposition header), and align filename extension handling with batch jobs ([#548](https://github.com/Open-EO/openeo-python-driver/issues/548))
 - Support extraction of default job options from predefined (and "custom") processes, and do basic merging ([#483](https://github.com/Open-EO/openeo-python-driver/issues/483))
 - `load_stac` with STAC metadata where "bands" dimension is erroneously missing: "bands" dimension will still be forced in dry run if "bands" argument is given to `load_stac` ([Open-EO/openeo-geopyspark-driver#1907](https://github.com/Open-EO/openeo-geopyspark-driver/issues/1907))
-
+- `export_workspace`: support not merging `derived_from` links into a STAC Collection. ([Open-EO/openeo-geopyspark-driver#1922](https://github.com/Open-EO/openeo-geopyspark-driver/issues/1922))
 
 ## 0.141.0
 

openeo_driver/_version.py

@@ -1 +1 @@
-__version__ = "0.142.0a4"
+__version__ = "0.142.0a5"

openeo_driver/workspace.py

@@ -2,6 +2,7 @@
 import logging
 import os.path
 import shutil
+from dataclasses import dataclass
 from pathlib import Path, PurePath
 from typing import Optional, Union
 from urllib.parse import urlparse
@@ -15,6 +16,11 @@
 _log = logging.getLogger(__name__)
 
 
+@dataclass(frozen=True)
+class MergeCollectionMetadataOptions:
+    merge_derived_from_links: bool = True
+
+
 class Workspace(abc.ABC):
     @abc.abstractmethod
     def import_file(self, common_path: Union[str, Path], file: Path, merge: str, remove_original: bool = False) -> str:
@@ -27,13 +33,20 @@ def import_object(
         raise NotImplementedError
 
     @abc.abstractmethod
-    def merge(self, stac_resource: STACObject, target: PurePath, remove_original: bool = False) -> STACObject:
+    def merge(
+        self,
+        stac_resource: STACObject,
+        target: PurePath,
+        remove_original: bool = False,
+        merge_collection_metadata_options: MergeCollectionMetadataOptions = MergeCollectionMetadataOptions(),
+    ) -> STACObject:
         """
         Merges a STAC resource, its children and their assets into this workspace at the given path,
         possibly removing the original assets.
         :param stac_resource: a STAC resource, typically a Collection
         :param target: a path identifier to a STAC resource to merge the given STAC resource into
         :param remove_original: remove the original assets?
+        :param merge_collection_metadata_options: options for merging collection metadata
         """
         raise NotImplementedError
 
@@ -64,10 +77,16 @@ def import_file(self, common_path: Union[str, Path], file: Path, merge: str, rem
         _log.debug(f"{'moved' if remove_original else 'copied'} {file_absolute} to {target_directory}")
         return f"file:{target_directory / file.name}"
 
-    def import_object(self, common_path: str, s3_uri: str, merge: str, remove_original: bool = False):
+    def import_object(self, common_path: Union[str, Path], s3_uri: str, merge: str, remove_original: bool = False):
         raise NotImplementedError(f"importing objects is not supported yet")
 
-    def merge(self, stac_resource: STACObject, target: PurePath, remove_original: bool = False) -> STACObject:
+    def merge(
+        self,
+        stac_resource: STACObject,
+        target: PurePath,
+        remove_original: bool = False,
+        merge_collection_metadata_options: MergeCollectionMetadataOptions = MergeCollectionMetadataOptions(),
+    ) -> STACObject:
         stac_resource = stac_resource.full_copy()
 
         target = self.root_directory / os.path.normpath(target).lstrip("/")
@@ -77,6 +96,8 @@ def merge(self, stac_resource: STACObject, target: PurePath, remove_original: bo
 
         if isinstance(stac_resource, Catalog):  # includes Collection
             new_collection = stac_resource
+            if not merge_collection_metadata_options.merge_derived_from_links:
+                new_collection.clear_links(rel=RelType.DERIVED_FROM)
 
             existing_collection = None
             try:
@@ -121,6 +142,7 @@ def replace_asset_href(asset: Asset, src_collection_path: Path) -> Asset:
                 new_collection = new_collection.map_assets(
                     lambda _, asset: replace_asset_href(asset, src_collection_path=new_collection_path)
                 )
+
                 new_collection.save(CatalogType.SELF_CONTAINED)
 
                 for new_item in new_collection.get_items():

tests/test_workspace.py

@@ -9,7 +9,7 @@
 import pytest
 
 from openeo_driver.util.date_math import now_utc
-from openeo_driver.workspace import DiskWorkspace
+from openeo_driver.workspace import DiskWorkspace, MergeCollectionMetadataOptions
 
 
 @pytest.mark.parametrize("merge", [
@@ -160,7 +160,8 @@ def test_merge_catalog_from_disk_new(tmp_path):
                 assert f.read() == "catalog-asset\n"
 
 
-def test_merge_from_disk_into_existing(tmp_path):
+@pytest.mark.parametrize("merge_derived_from_links", [True, False])
+def test_merge_from_disk_into_existing(tmp_path, merge_derived_from_links: bool):
     existing_collection = _collection(
         root_path=tmp_path / "src" / "existing_collection",
         collection_id="existing_collection",
@@ -219,10 +220,21 @@ def test_merge_from_disk_into_existing(tmp_path):
     )
 
     target = Path("path") / "to" / "collection.json"
+    merge_collection_metadata_options = MergeCollectionMetadataOptions(
+        merge_derived_from_links=merge_derived_from_links
+    )
 
     workspace = DiskWorkspace(root_directory=tmp_path)
-    workspace.merge(stac_resource=existing_collection, target=target)
-    imported_collection = workspace.merge(stac_resource=new_collection, target=target)
+    workspace.merge(
+        stac_resource=existing_collection,
+        target=target,
+        merge_collection_metadata_options=merge_collection_metadata_options,
+    )
+    imported_collection = workspace.merge(
+        stac_resource=new_collection,
+        target=target,
+        merge_collection_metadata_options=merge_collection_metadata_options,
+    )
 
     workspace_dir = (workspace.root_directory / target).parent
     assert _paths_relative_to(workspace_dir) == {
@@ -260,16 +272,20 @@ def test_merge_from_disk_into_existing(tmp_path):
          dt.datetime.fromisoformat("2024-11-04T00:00:00+00:00")]
     ]
 
-    assert [link.to_dict() for link in exported_collection.get_links(rel=RelType.DERIVED_FROM)] == [
-        {
-            "rel": "derived_from",
-            "href": "https://src.test/asset1.tif",
-        },
-        {
-            "rel": "derived_from",
-            "href": "https://src.test/asset2.tif",
-        },
-    ]
+    assert [link.to_dict() for link in exported_collection.get_links(rel=RelType.DERIVED_FROM)] == (
+        [
+            {
+                "rel": "derived_from",
+                "href": "https://src.test/asset1.tif",
+            },
+            {
+                "rel": "derived_from",
+                "href": "https://src.test/asset2.tif",
+            },
+        ]
+        if merge_derived_from_links
+        else []
+    )
 
     for item in exported_collection.get_items():
         for asset in item.get_assets().values():