feat: update dataset catalog with version

.github/workflows/codeql-analysis.yml

@@ -25,7 +25,7 @@ jobs:
           python -m pip install --upgrade pip
           python -m pip install poetry
           python -m poetry install \
-              --extras='cdk check_files_checksums check_stac_metadata content_iterator datasets dataset_versions import_dataset import_status populate_catalog validation_summary'
+              --extras='cdk check_files_checksums check_stac_metadata content_iterator datasets dataset_versions import_dataset import_status populate_catalog update_dataset_catalog validation_summary'
           echo "CODEQL_PYTHON=$(python -m poetry run which python)" >> $GITHUB_ENV
 
         # Initializes the CodeQL tools for scanning.

.github/workflows/deploy.yml

@@ -59,7 +59,7 @@ jobs:
           python -m pip install --upgrade pip
           python -m pip install poetry
           python -m poetry install \
-              --extras='cdk check_files_checksums check_stac_metadata content_iterator datasets dataset_versions import_dataset import_status populate_catalog validation_summary'
+              --extras='cdk check_files_checksums check_stac_metadata content_iterator datasets dataset_versions import_dataset import_status populate_catalog update_dataset_catalog validation_summary'
 
       - name: Print CDK version
         run: poetry run cdk --version

.github/workflows/prod-upgrade-deploy-test.yml

@@ -80,7 +80,7 @@ jobs:
           python -m pip install --upgrade pip
           python -m pip install poetry
           python -m poetry install \
-              --extras='cdk check_files_checksums check_stac_metadata content_iterator datasets dataset_versions import_dataset import_status populate_catalog validation_summary'
+              --extras='cdk check_files_checksums check_stac_metadata content_iterator datasets dataset_versions import_dataset import_status populate_catalog update_dataset_catalog validation_summary'
 
       - name: Print CDK version
         run: poetry run cdk --version
@@ -121,7 +121,7 @@ jobs:
           python -m pip install --upgrade pip
           python -m pip install poetry
           python -m poetry install \
-              --extras='cdk check_files_checksums check_stac_metadata content_iterator datasets dataset_versions import_dataset import_status populate_catalog validation_summary' \
+              --extras='cdk check_files_checksums check_stac_metadata content_iterator datasets dataset_versions import_dataset import_status populate_catalog update_dataset_catalog validation_summary' \
               --remove-untracked
 
       # deployment

.github/workflows/test.yml

@@ -37,7 +37,7 @@ jobs:
           python -m pip install --upgrade pip
           python -m pip install poetry
           python -m poetry install \
-              --extras='cdk check_files_checksums check_stac_metadata content_iterator datasets dataset_versions import_dataset import_status populate_catalog validation_summary'
+              --extras='cdk check_files_checksums check_stac_metadata content_iterator datasets dataset_versions import_dataset import_status populate_catalog update_dataset_catalog validation_summary'
 
       - name: Check all commit messages in Pull Request
         run: >
@@ -101,7 +101,7 @@ jobs:
           python -m pip install --upgrade pip
           python -m pip install poetry
           python -m poetry install \
-              --extras='cdk check_files_checksums check_stac_metadata content_iterator datasets dataset_versions import_dataset import_status populate_catalog validation_summary'
+              --extras='cdk check_files_checksums check_stac_metadata content_iterator datasets dataset_versions import_dataset import_status populate_catalog update_dataset_catalog validation_summary'
 
       - name: Print CDK version
         run: poetry run cdk --version

backend/api_keys.py

@@ -1,3 +1,5 @@
 MESSAGE_KEY = "message"
 STATUS_KEY = "status"
 SUCCESS_KEY = "success"
+
+EVENT_KEY = "event"

backend/check_stac_metadata/task.py

@@ -4,6 +4,7 @@
 from botocore.response import StreamingBody  # type: ignore[import]
 from jsonschema import ValidationError, validate  # type: ignore[import]
 
+from ..api_keys import EVENT_KEY
 from ..error_response_keys import ERROR_KEY, ERROR_MESSAGE_KEY
 from ..log import set_up_logging
 from ..models import DATASET_ID_PREFIX, DB_KEY_SEPARATOR, VERSION_ID_PREFIX
@@ -26,7 +27,7 @@ def s3_url_reader(url: str) -> StreamingBody:
 
 def lambda_handler(event: JsonObject, _context: bytes) -> JsonObject:
 
-    LOGGER.debug(dumps({"event": event}))
+    LOGGER.debug(dumps({EVENT_KEY: event}))
 
     # validate input
     try:

backend/check_stac_metadata/utils.py

@@ -15,12 +15,12 @@
 from ..s3 import S3_URL_PREFIX
 from ..stac_format import (
     STAC_ASSETS_KEY,
-    STAC_CATALOG_TYPE,
-    STAC_COLLECTION_TYPE,
     STAC_FILE_CHECKSUM_KEY,
     STAC_HREF_KEY,
-    STAC_ITEM_TYPE,
     STAC_LINKS_KEY,
+    STAC_TYPE_CATALOG,
+    STAC_TYPE_COLLECTION,
+    STAC_TYPE_ITEM,
     STAC_TYPE_KEY,
 )
 from ..types import JsonObject
@@ -41,9 +41,9 @@
         Type[STACItemSchemaValidator],
     ],
 ] = {
-    STAC_COLLECTION_TYPE: STACCollectionSchemaValidator,
-    STAC_CATALOG_TYPE: STACCatalogSchemaValidator,
-    STAC_ITEM_TYPE: STACItemSchemaValidator,
+    STAC_TYPE_COLLECTION: STACCollectionSchemaValidator,
+    STAC_TYPE_CATALOG: STACCatalogSchemaValidator,
+    STAC_TYPE_ITEM: STACItemSchemaValidator,
 }
 
 PROCESSING_ASSET_ASSET_KEY = "asset"

backend/datasets/create.py

@@ -11,6 +11,14 @@
 from ..parameter_store import ParameterName, get_param
 from ..pystac_io_methods import write_method
 from ..resources import ResourceName
+from ..s3 import S3_URL_PREFIX
+from ..sqs_message_attributes import (
+    DATA_TYPE_KEY,
+    DATA_TYPE_STRING,
+    MESSAGE_ATTRIBUTE_TYPE_KEY,
+    MESSAGE_ATTRIBUTE_TYPE_ROOT,
+    STRING_VALUE_KEY,
+)
 from ..stac_format import STAC_DESCRIPTION_KEY, STAC_ID_KEY, STAC_TITLE_KEY
 from ..types import JsonObject
 
@@ -60,14 +68,22 @@ def create_dataset(body: JsonObject) -> JsonObject:
         catalog_type=CatalogType.SELF_CONTAINED,
     )
     dataset_catalog.normalize_hrefs(
-        f"s3://{ResourceName.STORAGE_BUCKET_NAME.value}/{dataset.dataset_prefix}"
+        f"{S3_URL_PREFIX}{ResourceName.STORAGE_BUCKET_NAME.value}/{dataset.dataset_prefix}"
     )
     dataset_catalog.save()
 
     # add reference to root catalog
     SQS_RESOURCE.get_queue_by_name(
-        QueueName=get_param(ParameterName.ROOT_CATALOG_MESSAGE_QUEUE_NAME)
-    ).send_message(MessageBody=dataset.dataset_prefix)
+        QueueName=get_param(ParameterName.UPDATE_CATALOG_MESSAGE_QUEUE_NAME)
+    ).send_message(
+        MessageBody=dataset.dataset_prefix,
+        MessageAttributes={
+            MESSAGE_ATTRIBUTE_TYPE_KEY: {
+                STRING_VALUE_KEY: MESSAGE_ATTRIBUTE_TYPE_ROOT,
+                DATA_TYPE_KEY: DATA_TYPE_STRING,
+            }
+        },
+    )
 
     # return response
     resp_body = dataset.as_dict()

backend/import_dataset/task.py

@@ -9,6 +9,7 @@
 from jsonschema import ValidationError, validate  # type: ignore[import]
 from smart_open import open as smart_open  # type: ignore[import]
 
+from ..api_keys import EVENT_KEY
 from ..error_response_keys import ERROR_KEY, ERROR_MESSAGE_KEY
 from ..import_dataset_keys import NEW_KEY_KEY, ORIGINAL_KEY_KEY, TARGET_BUCKET_NAME_KEY
 from ..import_file_batch_job_id_keys import ASSET_JOB_ID_KEY, METADATA_JOB_ID_KEY
@@ -70,8 +71,6 @@
 
 S3_BATCH_COPY_ROLE_ARN = get_param(ParameterName.PROCESSING_IMPORT_DATASET_ROLE_ARN)
 
-EVENT_KEY = "event"
-
 JOB_MANIFEST_FORMAT: JobManifestFormatType = "S3BatchOperations_CSV_20180820"
 JOB_MANIFEST_FIELD_NAMES: List[JobManifestFieldNameType] = ["Bucket", "Key"]
 JOB_REPORT_FORMAT: JobReportFormatType = "Report_CSV_20180820"

backend/parameter_store.py

@@ -26,7 +26,7 @@ def _generate_next_value_(  # type: ignore[misc,override] # pylint:disable=no-se
     PROCESSING_IMPORT_ASSET_FILE_FUNCTION_TASK_ARN = auto()
     PROCESSING_IMPORT_DATASET_ROLE_ARN = auto()
     PROCESSING_IMPORT_METADATA_FILE_FUNCTION_TASK_ARN = auto()
-    ROOT_CATALOG_MESSAGE_QUEUE_NAME = auto()
+    UPDATE_CATALOG_MESSAGE_QUEUE_NAME = auto()
     STORAGE_DATASETS_TABLE_NAME = auto()
     STORAGE_VALIDATION_RESULTS_TABLE_NAME = auto()
 

backend/populate_catalog/task.py

@@ -1,10 +1,23 @@
+from json import dumps
+from os.path import join
+from urllib.parse import urlparse
+
 import boto3
-from pystac import STAC_IO, Catalog, CatalogType  # type: ignore[import]
+from pystac import STAC_IO, Catalog, CatalogType, Collection, Item  # type: ignore[import]
+from pystac.layout import HrefLayoutStrategy  # type: ignore[import]
 
+from ..api_keys import EVENT_KEY
 from ..api_responses import BODY_KEY
+from ..log import set_up_logging
 from ..pystac_io_methods import read_method, write_method
 from ..resources import ResourceName
 from ..s3 import S3_URL_PREFIX
+from ..sqs_message_attributes import (
+    MESSAGE_ATTRIBUTE_TYPE_DATASET,
+    MESSAGE_ATTRIBUTE_TYPE_KEY,
+    MESSAGE_ATTRIBUTE_TYPE_ROOT,
+    STRING_VALUE_KEY_LOWER,
+)
 from ..types import JsonObject
 
 STAC_IO.write_text_method = write_method
@@ -18,11 +31,78 @@
 CATALOG_KEY = "catalog.json"
 CONTENTS_KEY = "Contents"
 RECORDS_KEY = "Records"
+MESSAGE_ATTRIBUTES_KEY = "messageAttributes"
+
+LOGGER = set_up_logging(__name__)
 
 
 def lambda_handler(event: JsonObject, _context: bytes) -> JsonObject:
     """Main Lambda entry point."""
 
+    LOGGER.debug(dumps({EVENT_KEY: event}))
+
+    for message in event[RECORDS_KEY]:
+        if (
+            message[MESSAGE_ATTRIBUTES_KEY][MESSAGE_ATTRIBUTE_TYPE_KEY][STRING_VALUE_KEY_LOWER]
+            == MESSAGE_ATTRIBUTE_TYPE_ROOT
+        ):
+            handle_root(message[BODY_KEY])
+        elif (
+            message[MESSAGE_ATTRIBUTES_KEY][MESSAGE_ATTRIBUTE_TYPE_KEY][STRING_VALUE_KEY_LOWER]
+            == MESSAGE_ATTRIBUTE_TYPE_DATASET
+        ):
+            handle_dataset(message[BODY_KEY])
+        else:
+            raise UnhandledSQSMessageException("Unhandled SQS message type")
+
+    return {}
+
+
+class UnhandledSQSMessageException(Exception):
+    pass
+
+
+class GeostoreSTACLayoutStrategy(HrefLayoutStrategy):
+    def get_catalog_href(self, cat: Catalog, parent_dir: str, is_root: bool) -> str:
+        original_path = urlparse(cat.get_self_href()).path.rsplit("/", maxsplit=2)
+        if is_root:
+            cat_root = parent_dir
+        else:
+            cat_root = join(parent_dir, original_path[-2])
+
+        return join(cat_root, original_path[-1])
+
+    def get_collection_href(self, col: Collection, parent_dir: str, is_root: bool) -> str:
+        original_path = urlparse(col.get_self_href()).path.rsplit("/", maxsplit=2)
+        assert not is_root
+        return join(parent_dir, *original_path[-2:])
+
+    def get_item_href(self, item: Item, parent_dir: str) -> str:
+        original_path = item.get_self_href().split("/")
+        return join(parent_dir, original_path[-1])
+
+
+def handle_dataset(version_metadata_key: str) -> None:
+    """Handle writing a new dataset version to the dataset catalog"""
+    storage_bucket_path = f"{S3_URL_PREFIX}{ResourceName.STORAGE_BUCKET_NAME.value}"
+    dataset_prefix = version_metadata_key.split("/", maxsplit=1)[0]
+
+    dataset_catalog = Catalog.from_file(f"{storage_bucket_path}/{dataset_prefix}/{CATALOG_KEY}")
+
+    dataset_version_metadata = STAC_IO.read_stac_object(
+        f"{storage_bucket_path}/{version_metadata_key}"
+    )
+
+    dataset_catalog.add_child(dataset_version_metadata, strategy=GeostoreSTACLayoutStrategy())
+
+    dataset_catalog.normalize_hrefs(
+        f"{storage_bucket_path}/{dataset_prefix}", strategy=GeostoreSTACLayoutStrategy()
+    )
+    dataset_catalog.save(catalog_type=CatalogType.SELF_CONTAINED)
+
+
+def handle_root(dataset_prefix: str) -> None:
+    """Handle writing a new dataset to the root catalog"""
     results = S3_CLIENT.list_objects(
         Bucket=ResourceName.STORAGE_BUCKET_NAME.value, Prefix=CATALOG_KEY
     )
@@ -40,14 +120,17 @@ def lambda_handler(event: JsonObject, _context: bytes) -> JsonObject:
             description=ROOT_CATALOG_DESCRIPTION,
             catalog_type=CatalogType.SELF_CONTAINED,
         )
+        root_catalog.set_self_href(
+            f"{S3_URL_PREFIX}{ResourceName.STORAGE_BUCKET_NAME.value}/{CATALOG_KEY}"
+        )
 
-    for record in event[RECORDS_KEY]:
-        dataset_path = f"{S3_URL_PREFIX}{ResourceName.STORAGE_BUCKET_NAME.value}/{record[BODY_KEY]}"
-        dataset_catalog = Catalog.from_file(f"{dataset_path}/{CATALOG_KEY}")
-
-        root_catalog.add_child(dataset_catalog)
-        root_catalog.normalize_hrefs(f"{S3_URL_PREFIX}{ResourceName.STORAGE_BUCKET_NAME.value}")
+    dataset_path = f"{S3_URL_PREFIX}{ResourceName.STORAGE_BUCKET_NAME.value}/{dataset_prefix}"
+    dataset_catalog = Catalog.from_file(f"{dataset_path}/{CATALOG_KEY}")
 
-        root_catalog.save(catalog_type=CatalogType.SELF_CONTAINED)
+    root_catalog.add_child(dataset_catalog, strategy=GeostoreSTACLayoutStrategy())
+    root_catalog.normalize_hrefs(
+        f"{S3_URL_PREFIX}{ResourceName.STORAGE_BUCKET_NAME.value}",
+        strategy=GeostoreSTACLayoutStrategy(),
+    )
 
-    return {}
+    root_catalog.save(catalog_type=CatalogType.SELF_CONTAINED)

backend/sqs_message_attributes.py

@@ -0,0 +1,16 @@
+def decapitalize(key: str) -> str:
+    """
+    This method will be used to lower case the first character of SQS
+    message attributes being received by Lambda to resolve inconsistencies.
+    Issue outlined here: https://github.com/boto/boto3/issues/2582
+    """
+    return f"{key[:1].lower()}{key[1:]}"
+
+
+MESSAGE_ATTRIBUTE_TYPE_KEY = "type"
+MESSAGE_ATTRIBUTE_TYPE_ROOT = "root"
+MESSAGE_ATTRIBUTE_TYPE_DATASET = "dataset"
+DATA_TYPE_KEY = "DataType"
+DATA_TYPE_STRING = "String"
+STRING_VALUE_KEY = "StringValue"
+STRING_VALUE_KEY_LOWER = decapitalize(STRING_VALUE_KEY)

backend/stac_format.py

@@ -10,16 +10,20 @@
 STAC_HREF_KEY = "href"
 STAC_ID_KEY = "id"
 STAC_LICENSE_KEY = "license"
+STAC_LINKS_KEY = "links"
+STAC_MEDIA_TYPE_GEOJSON = "application/geo+json"
+STAC_MEDIA_TYPE_JSON = "application/json"
 STAC_PROPERTIES_DATETIME_KEY = "datetime"
 STAC_PROPERTIES_KEY = "properties"
+STAC_REL_CHILD = "child"
+STAC_REL_ITEM = "item"
+STAC_REL_KEY = "rel"
+STAC_REL_PARENT = "parent"
+STAC_REL_ROOT = "root"
+STAC_REL_SELF = "self"
 STAC_TITLE_KEY = "title"
+STAC_TYPE_CATALOG = "Catalog"
+STAC_TYPE_COLLECTION = "Collection"
+STAC_TYPE_ITEM = "Feature"
 STAC_TYPE_KEY = "type"
 STAC_VERSION_KEY = "stac_version"
-STAC_LINKS_KEY = "links"
-STAC_REL_KEY = "rel"
-STAC_REL_ROOT = "root"
-STAC_REL_CHILD = "child"
-
-STAC_COLLECTION_TYPE = "Collection"
-STAC_ITEM_TYPE = "Feature"
-STAC_CATALOG_TYPE = "Catalog"