Added 'add' command to add an item to a catalog/collection

src/stactools/cli/__init__.py

@@ -9,7 +9,7 @@ def register_plugin(registry):
     # Register subcommands
 
     from stactools.cli.commands import (copy, info, layout, merge, migrate,
-                                        version)
+                                        version, add)
 
     registry.register_subcommand(copy.create_copy_command)
     registry.register_subcommand(copy.create_move_assets_command)
@@ -18,6 +18,7 @@ def register_plugin(registry):
     registry.register_subcommand(layout.create_layout_command)
     registry.register_subcommand(merge.create_merge_command)
     registry.register_subcommand(version.create_version_command)
+    registry.register_subcommand(add.create_add_command)
 
     # TODO
     # registry.register_subcommand(migrate.create_migrate_command)

src/stactools/cli/commands/add.py

@@ -0,0 +1,41 @@
+import click
+import pystac
+
+from stactools.core import add_item
+
+
+def add(source_item, target_catalog, collection_id=None, move_assets=False):
+    source = pystac.read_file(source_item)
+    target = pystac.read_file(target_catalog)
+
+    if collection_id is not None:
+        target = target.get_child(collection_id, recursive=True)
+        if target is None:
+            raise click.BadOptionUsage(
+                'collection',
+                'A collection with ID {} does not exist in {}'.format(
+                    collection_id, target_catalog))
+
+    add_item(source, target, move_assets)
+
+    target.save()
+
+
+def create_add_command(cli):
+    @cli.command('add', short_help='Add an item to a catalog/collection.')
+    @click.argument('source_item')
+    @click.argument('target_catalog')
+    @click.option('--collection',
+                  help=("The collection ID to add to. If not set, will "
+                        "add to the root catalog or collection."))
+    @click.option('-a',
+                  '--move-assets',
+                  is_flag=True,
+                  help='Move assets to the target catalog Item locations.')
+    def add_command(source_item, target_catalog, collection, move_assets):
+        add(source_item,
+            target_catalog,
+            collection_id=collection,
+            move_assets=move_assets)
+
+    return add_command

src/stactools/core/__init__.py

@@ -5,5 +5,6 @@
                                  move_all_assets, copy_catalog)
 from stactools.core.layout import layout_catalog
 from stactools.core.merge import (merge_items, merge_all_items)
+from stactools.core.add import add_item
 
 __version__ = "0.2.1a2"

src/stactools/core/add.py

@@ -0,0 +1,41 @@
+import os
+
+import pystac
+from pystac.layout import BestPracticesLayoutStrategy
+
+from stactools.core.copy import move_assets as do_move_assets
+
+
+def add_item(source_item, target_catalog, move_assets=False):
+    """Add a item into a catalog.
+
+    Args:
+        source_item (pystac.Item): The Item that will be added.
+            This item is not mutated in this operation.
+        target_catalog (pystac.Item): The destination catalog.
+            This catalog will be mutated in this operation.
+        move_assets (bool): If true, move the asset files alongside the target item.
+    """
+
+    target_item_ids = [item.id for item in target_catalog.get_all_items()]
+    if source_item.id in target_item_ids:
+        raise ValueError(
+            f'An item with ID {source_item.id} already exists in the target catalog'
+        )
+    parent_dir = os.path.dirname(target_catalog.get_self_href())
+    layout_strategy = BestPracticesLayoutStrategy()
+    item_copy = source_item.clone()
+    item_copy.set_self_href(
+        layout_strategy.get_item_href(item_copy, parent_dir))
+    target_catalog.add_item(item_copy)
+
+    if isinstance(target_catalog, pystac.Collection):
+        item_copy.set_collection(target_catalog)
+    else:
+        item_copy.set_collection(None)
+
+    if move_assets:
+        do_move_assets(item_copy, copy=False)
+
+    if target_catalog.STAC_OBJECT_TYPE == pystac.STACObjectType.COLLECTION:
+        target_catalog.update_extent_from_items()

tests/cli/commands/test_add.py

@@ -0,0 +1,90 @@
+from tempfile import TemporaryDirectory
+
+import pystac
+from stactools.core import move_all_assets
+from stactools.cli.commands.add import create_add_command
+from stactools.testing import CliTestCase
+from .test_cases import TestCases
+
+
+def create_temp_catalog_copy(tmp_dir):
+    col = TestCases.planet_disaster()
+    col.normalize_hrefs(tmp_dir)
+    col.save(catalog_type=pystac.CatalogType.SELF_CONTAINED)
+    move_all_assets(col, copy=True)
+    col.save()
+
+    return col
+
+
+class AddTest(CliTestCase):
+    def create_subcommand_functions(self):
+        return [create_add_command]
+
+    def test_add_item(self):
+        catalog = TestCases.test_case_1()
+        subcatalog = list(list(catalog.get_children())[0].get_children())[0]
+        item = list(subcatalog.get_all_items())[0]
+        item_path = item.get_self_href()
+        with TemporaryDirectory() as tmp_dir:
+            target_catalog = create_temp_catalog_copy(tmp_dir)
+
+            items = list(target_catalog.get_all_items())
+            self.assertEqual(len(items), 5)
+
+            cmd = ["add", item_path, target_catalog.get_self_href()]
+
+            self.run_command(cmd)
+
+            target_col = pystac.read_file(target_catalog.get_self_href())
+            items = list(target_col.get_all_items())
+            self.assertEqual(len(items), 6)
+
+    def test_add_item_to_specific_collection(self):
+        catalog = TestCases.test_case_1()
+        subcatalog = list(list(catalog.get_children())[0].get_children())[0]
+        item = list(subcatalog.get_all_items())[0]
+        item_path = item.get_self_href()
+        with TemporaryDirectory() as tmp_dir:
+            target_catalog = create_temp_catalog_copy(tmp_dir)
+
+            items = list(target_catalog.get_all_items())
+            self.assertEqual(len(items), 5)
+
+            cmd = [
+                "add",
+                item_path,
+                target_catalog.get_self_href(),
+                "--collection",
+                "hurricane-harvey",
+            ]
+
+            res = self.run_command(cmd)
+            self.assertEqual(res.exit_code, 0)
+
+            target_col = pystac.read_file(target_catalog.get_self_href())
+            items = list(target_col.get_all_items())
+            self.assertEqual(len(items), 6)
+
+    def test_add_item_to_missing_collection(self):
+        catalog = TestCases.test_case_1()
+        subcatalog = list(list(catalog.get_children())[0].get_children())[0]
+        item = list(subcatalog.get_all_items())[0]
+        item_path = item.get_self_href()
+        with TemporaryDirectory() as tmp_dir:
+            target_catalog = create_temp_catalog_copy(tmp_dir)
+
+            items = list(target_catalog.get_all_items())
+            self.assertEqual(len(items), 5)
+
+            cmd = [
+                "add",
+                item_path,
+                target_catalog.get_self_href(),
+                "--collection",
+                "WRONG",
+            ]
+
+            res = self.run_command(cmd)
+            self.assertEqual(res.exit_code, 2)
+            self.assertTrue(" A collection with ID WRONG does not exist" in res.output)