fix: sanitize collection ids

cdk/constructs/DpsStacItemGenerator/runtime/pyproject.toml

@@ -11,6 +11,7 @@ dependencies = [
     "pydantic>=2.11.0",
     "pystac[validation]>=1.13.0",
     "stac-pydantic>=3.2.0",
+    "python-slugify==8.0.4",
 ]
 
 [dependency-groups]

cdk/constructs/DpsStacItemGenerator/runtime/src/dps_stac_item_generator/item.py

@@ -10,6 +10,7 @@
 from pystac import Link
 from pystac.stac_io import DefaultStacIO, StacIO
 from stac_pydantic.item import Item
+from slugify import slugify
 
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
@@ -94,7 +95,7 @@ def get_stac_items(catalog_json_key: str) -> Generator[Item, Any, Any]:
             f"could not locate the .met.json file with the DPS job outputs in {job_output_prefix}"
         )
 
-    collection_id = COLLECTION_ID_FORMAT.format(**job_metadata)
+    collection_id = slugify(COLLECTION_ID_FORMAT.format(**job_metadata), regex_pattern=r'[/\?#%& ]+')
 
     catalog = pystac.Catalog.from_file(catalog_json_key)
     catalog.make_all_asset_hrefs_absolute()

cdk/constructs/DpsStacItemGenerator/runtime/tests/test_item.py

@@ -300,3 +300,25 @@ def test_get_stac_items_invalid_catalog_json(self, mock_job_metadata):
                 Exception, match="Failed to parse catalog.json: invalid format"
             ):
                 list(get_stac_items(catalog_s3_key))
+
+    def test_santitize_collection_id(self, mock_catalog, mock_job_metadata):
+        """Test that collection ID is sanitized correctly."""
+        catalog_s3_key = "s3://test-bucket/2023/01/15/10/30/45/123456/catalog.json"
+        mock_job_metadata["username"] = "user/name"
+        mock_job_metadata["algorithm_name"] = "algo?name"
+        expected_collection_id = "user-name__algo-name__0.1__test"
+
+        with (
+            patch(
+                "dps_stac_item_generator.item.pystac.Catalog.from_file",
+                return_value=mock_catalog,
+            ),
+            patch(
+                "dps_stac_item_generator.item.load_met_json",
+                return_value=mock_job_metadata,
+            ),
+        ):
+            items = list(get_stac_items(catalog_s3_key))
+
+            for item in items:
+                assert item.collection == expected_collection_id