In 1101 base workflow

README.md

@@ -22,6 +22,7 @@ Description of the app
 SENTRY_DSN=### If set to a valid Sentry DSN, enables Sentry exception monitoring. This is not needed for local development.
 WORKSPACE=### Set to `dev` for local development, this will be set to `stage` and `prod` in those environments by Terraform.
 AWS_REGION_NAME=### Default AWS region.
+DSS_INPUT_QUEUE=### The DSS SQS input queue to which submission messages are sent.
 ```
 
 ### Optional

dsc/config.py

@@ -11,6 +11,7 @@ class Config:
         "WORKSPACE",
         "SENTRY_DSN",
         "AWS_REGION_NAME",
+        "DSS_INPUT_QUEUE",
     ]
 
     OPTIONAL_ENV_VARS: Iterable[str] = ["LOG_LEVEL"]

dsc/exceptions.py

@@ -1,2 +1,10 @@
+class InvalidDSpaceMetadataError(Exception):
+    pass
+
+
 class InvalidSQSMessageError(Exception):
     pass
+
+
+class ItemMetadatMissingRequiredFieldError(Exception):
+    pass

dsc/item_submission.py

@@ -0,0 +1,30 @@
+import json
+import logging
+from dataclasses import dataclass
+from typing import Any
+
+from dsc.utilities.aws.s3 import S3Client
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ItemSubmission:
+    """A class to store the required values for a DSpace submission."""
+
+    dspace_metadata: dict[str, Any]
+    bitstream_uris: list[str]
+    metadata_s3_key: str
+    metadata_uri: str = ""
+
+    def upload_dspace_metadata(self, bucket: str) -> None:
+        """Upload DSpace metadata to S3 using the specified bucket and keyname.
+
+        Args:
+            bucket: The S3 bucket for uploading the item metadata file.
+        """
+        s3_client = S3Client()
+        s3_client.put_file(json.dumps(self.dspace_metadata), bucket, self.metadata_s3_key)
+        metadata_uri = f"s3://{bucket}/{self.metadata_s3_key}"
+        logger.info(f"Metadata uploaded to S3: {metadata_uri}")
+        self.metadata_uri = metadata_uri

dsc/workflows/__init__.py

@@ -0,0 +1,10 @@
+"""dsc.workflows.
+
+All primary functions used by CLI are importable from here.
+"""
+
+from dsc.workflows.base import BaseWorkflow
+
+__all__ = [
+    "BaseWorkflow",
+]