In 1101 base workflow

dsc/base.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import logging
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, Any, final
+
+from dsc.item_submission import ItemSubmission
+from dsc.utilities.aws.s3 import S3Client
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+logger = logging.getLogger(__name__)
+
+
+class BaseWorkflow(ABC):
+    """A base workflow class from which other workflow classes are derived."""
+
+    def __init__(
+        self,
+        email_recipients: list[str],
+        metadata_mapping: dict,
+        s3_bucket: str,
+        s3_prefix: str | None,
+    ) -> None:
+        """Initialize base instance.
+
+        Args:
+            email_recipients: The email addresses to notify after runs of the workflow.
+            metadata_mapping: A mapping file for generating DSpace metadata from the
+            workflow's source metadata.
+            s3_bucket: The S3 bucket containing bitstream and metadata files for the
+            workflow.
+            s3_prefix: The S3 prefix used for objects in this workflow. This prefix does
+            NOT include the bucket name.
+        """
+        self.email_recipients: list[str] = email_recipients
+        self.metadata_mapping: dict = metadata_mapping
+        self.s3_bucket: str = s3_bucket
+        self.s3_prefix: str | None = s3_prefix
+
+    @final
+    def generate_submission_batch(self) -> Iterator[tuple[str, list[str]]]:
+        """Generate a batch of item submissions for the DSpace Submission Service.
+
+        MUST NOT be overridden by workflow subclasses.
+        """
+        s3_client = S3Client()
+        batch_metadata = self.get_batch_metadata()
+        for item_metadata in batch_metadata:
+            item_identifier = self.get_item_identifier(item_metadata)
+            logger.info(f"Processing submission for '{item_identifier}'")
+            metadata_keyname = f"{self.s3_prefix}/{item_identifier}_metadata.json"
+            item_submission = ItemSubmission(
+                source_metadata=item_metadata,
+                metadata_mapping=self.metadata_mapping,
+                s3_client=s3_client,
+                bitstream_uris=self.get_bitstream_uris(item_identifier),
+                metadata_keyname=metadata_keyname,
+            )
+            item_submission.generate_and_upload_dspace_metadata(self.s3_bucket)
+            yield item_submission.metadata_uri, item_submission.bitstream_uris
+
+    @abstractmethod
+    def get_batch_metadata(self) -> list:
+        """Get source metadata for the batch of items submissions.
+
+        MUST be overridden by workflow subclasses.
+        """
+
+    @abstractmethod
+    def get_item_identifier(self, item_metadata: Any) -> str:  # noqa: ANN401
+        """Get identifier for an item submission according to the workflow subclass.
+
+        MUST be overridden by workflow subclasses.
+
+        Args:
+            item_metadata: The item metadata from which the item identifier is extracted.
+        """
+
+    @abstractmethod
+    def get_bitstream_uris(self, item_identifier: str) -> list[str]:
+        """Get bitstreams for an item submission according to the workflow subclass.
+
+        MUST be overridden by workflow subclasses.
+
+        Args:
+            item_identifier: The identifier used for locating the item's bitstreams.
+        """
+
+    @abstractmethod
+    def process_deposit_results(self) -> list[str]:
+        """Process results generated by the deposit according to the workflow subclass.
+
+        MUST be overridden by workflow subclasses.
+        """

dsc/item_submission.py

@@ -0,0 +1,68 @@
+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."""
+
+    source_metadata: dict[str, Any]
+    metadata_mapping: dict[str, Any]
+    s3_client: S3Client
+    bitstream_uris: list[str]
+    metadata_keyname: str
+    metadata_uri: str = ""
+
+    def generate_and_upload_dspace_metadata(self, bucket: str) -> None:
+        """Generate DSpace metadata from the item's source metadata and upload it to S3.
+
+        Args:
+            bucket: The S3 bucket for uploading the item metadata file.
+        """
+        dspace_metadata = self.create_dspace_metadata()
+        self.s3_client.put_file(
+            json.dumps(dspace_metadata),
+            bucket,
+            self.metadata_keyname,
+        )
+        metadata_uri = f"s3://{bucket}/{self.metadata_keyname}"
+        logger.info(f"Metadata uploaded to S3: {metadata_uri}")
+        self.metadata_uri = metadata_uri
+
+    def create_dspace_metadata(self) -> dict[str, Any]:
+        """Create DSpace metadata from the item's source metadata."""
+        metadata_entries = []
+        for field_name, field_mapping in self.metadata_mapping.items():
+            if field_name not in ["item_identifier", "source_system_identifier"]:
+
+                field_value = self.source_metadata.get(field_mapping["source_field_name"])
+                if field_value:
+                    delimiter = field_mapping["delimiter"]
+                    language = field_mapping["language"]
+                    if delimiter:
+                        metadata_entries.extend(
+                            [
+                                {
+                                    "key": field_name,
+                                    "value": value,
+                                    "language": language,
+                                }
+                                for value in field_value.split(delimiter)
+                            ]
+                        )
+                    else:
+                        metadata_entries.append(
+                            {
+                                "key": field_name,
+                                "value": field_value,
+                                "language": language,
+                            }
+                        )
+
+        return {"metadata": metadata_entries}

tests/conftest.py

@@ -7,6 +7,7 @@
 from moto import mock_aws
 
 from dsc.config import Config
+from dsc.item_submission import ItemSubmission
 from dsc.utilities.aws.s3 import S3Client
 from dsc.utilities.aws.ses import SESClient
 from dsc.utilities.aws.sqs import SQSClient
@@ -22,15 +23,51 @@ def _test_env(monkeypatch):
 
 
 @pytest.fixture
-def config_instance() -> Config:
+def config_instance():
     return Config()
 
 
+@pytest.fixture
+def item_submission_instance(metadata_mapping, s3_client):
+    source_metadata = {"title": "Title", "contributor": "Author 1|Author 2"}
+    return ItemSubmission(
+        source_metadata=source_metadata,
+        metadata_mapping=metadata_mapping,
+        s3_client=s3_client,
+        bitstream_uris=[
+            "s3://dsc/workflow/folder/123_01.pdf",
+            "s3://dsc/workflow/folder/123_02.pdf",
+        ],
+        metadata_keyname="workflow/folder/123_metadata.json",
+    )
+
+
+@pytest.fixture
+def metadata_mapping():
+    return {
+        "item_identifier": {
+            "source_field_name": "item_identifier",
+            "language": None,
+            "delimiter": "",
+        },
+        "dc.title": {
+            "source_field_name": "title",
+            "language": "en_US",
+            "delimiter": "",
+        },
+        "dc.contributor": {
+            "source_field_name": "contributor",
+            "language": None,
+            "delimiter": "|",
+        },
+    }
+
+
 @pytest.fixture
 def mocked_s3(config_instance):
     with mock_aws():
         s3 = boto3.client("s3", region_name=config_instance.AWS_REGION_NAME)
-        s3.create_bucket(Bucket="awd")
+        s3.create_bucket(Bucket="dsc")
         yield s3
 
 
@@ -130,11 +167,11 @@ def submission_message_body():
         {
             "SubmissionSystem": "DSpace@MIT",
             "CollectionHandle": "123.4/5678",
-            "MetadataLocation": "s3://awd/10.1002-term.3131.json",
+            "MetadataLocation": "s3://dsc/10.1002-term.3131.json",
             "Files": [
                 {
                     "BitstreamName": "10.1002-term.3131.pdf",
-                    "FileLocation": "s3://awd/10.1002-term.3131.pdf",
+                    "FileLocation": "s3://dsc/10.1002-term.3131.pdf",
                     "BitstreamDescription": None,
                 }
             ],

tests/test_itemsubmission.py

@@ -0,0 +1,64 @@
+from http import HTTPStatus
+
+
+def test_itemsubmission_init_success(item_submission_instance):
+    assert item_submission_instance.source_metadata == {
+        "title": "Title",
+        "contributor": "Author 1|Author 2",
+    }
+    assert item_submission_instance.metadata_mapping == {
+        "dc.contributor": {
+            "delimiter": "|",
+            "language": None,
+            "source_field_name": "contributor",
+        },
+        "dc.title": {"delimiter": "", "language": "en_US", "source_field_name": "title"},
+        "item_identifier": {
+            "delimiter": "",
+            "language": None,
+            "source_field_name": "item_identifier",
+        },
+    }
+    assert item_submission_instance.bitstream_uris == [
+        "s3://dsc/workflow/folder/123_01.pdf",
+        "s3://dsc/workflow/folder/123_02.pdf",
+    ]
+    assert (
+        item_submission_instance.metadata_keyname == "workflow/folder/123_metadata.json"
+    )
+
+
+def test_generate_and_upload_dspace_metadata(
+    mocked_s3, item_submission_instance, s3_client
+):
+    item_submission_instance.generate_and_upload_dspace_metadata("dsc")
+    assert (
+        item_submission_instance.metadata_uri
+        == "s3://dsc/workflow/folder/123_metadata.json"
+    )
+    response = s3_client.client.get_object(
+        Bucket="dsc", Key="workflow/folder/123_metadata.json"
+    )
+    assert response["ResponseMetadata"]["HTTPStatusCode"] == HTTPStatus.OK
+
+
+def test_create_dspace_metadata(item_submission_instance):
+    assert item_submission_instance.create_dspace_metadata() == {
+        "metadata": [
+            {
+                "key": "dc.title",
+                "language": "en_US",
+                "value": "Title",
+            },
+            {
+                "key": "dc.contributor",
+                "language": None,
+                "value": "Author 1",
+            },
+            {
+                "key": "dc.contributor",
+                "language": None,
+                "value": "Author 2",
+            },
+        ]
+    }

tests/test_s3.py

@@ -7,60 +7,60 @@
 def test_s3_archive_file_in_bucket(mocked_s3, s3_client):
     s3_client.put_file(
         file_content="test1,test2,test3,test4",
-        bucket="awd",
+        bucket="dsc",
         key="test.csv",
     )
     s3_client.archive_file_with_new_key(
-        bucket="awd",
+        bucket="dsc",
         key="test.csv",
         archived_key_prefix="archived",
     )
     with pytest.raises(ClientError) as e:
-        response = s3_client.client.get_object(Bucket="awd", Key="test.csv")
+        response = s3_client.client.get_object(Bucket="dsc", Key="test.csv")
     assert (
         "An error occurred (NoSuchKey) when calling the GetObject operation: The"
         " specified key does not exist." in str(e.value)
     )
-    response = s3_client.client.get_object(Bucket="awd", Key="archived/test.csv")
+    response = s3_client.client.get_object(Bucket="dsc", Key="archived/test.csv")
     assert response["ResponseMetadata"]["HTTPStatusCode"] == HTTPStatus.OK
 
 
 def test_s3_put_file(mocked_s3, s3_client):
-    assert "Contents" not in s3_client.client.list_objects(Bucket="awd")
+    assert "Contents" not in s3_client.client.list_objects(Bucket="dsc")
     s3_client.put_file(
         file_content=str({"metadata": {"key": "dc.title", "value": "A Title"}}),
-        bucket="awd",
+        bucket="dsc",
         key="test.json",
     )
-    assert len(s3_client.client.list_objects(Bucket="awd")["Contents"]) == 1
+    assert len(s3_client.client.list_objects(Bucket="dsc")["Contents"]) == 1
     assert (
-        s3_client.client.list_objects(Bucket="awd")["Contents"][0]["Key"] == "test.json"
+        s3_client.client.list_objects(Bucket="dsc")["Contents"][0]["Key"] == "test.json"
     )
 
 
 def test_s3_get_files_iter_with_matching_csv(mocked_s3, s3_client):
     s3_client.put_file(
         file_content="test1,test2,test3,test4",
-        bucket="awd",
+        bucket="dsc",
         key="test.csv",
     )
     assert list(
         s3_client.get_files_iter(
-            bucket="awd", file_type="csv", excluded_key_prefix="archived"
+            bucket="dsc", file_type="csv", excluded_key_prefix="archived"
         )
     ) == ["test.csv"]
 
 
 def test_s3_get_files_iter_without_matching_csv(mocked_s3, s3_client):
     s3_client.put_file(
         file_content="test1,test2,test3,test4",
-        bucket="awd",
+        bucket="dsc",
         key="archived/test.csv",
     )
     assert (
         list(
             s3_client.get_files_iter(
-                bucket="awd", file_type="csv", excluded_key_prefix="archived"
+                bucket="dsc", file_type="csv", excluded_key_prefix="archived"
             )
         )
         == []

tests/test_sqs.py

@@ -17,9 +17,9 @@ def test_sqs_create_dss_message_body(sqs_client, submission_message_body):
     dss_message_body = sqs_client.create_dss_message_body(
         submission_system="DSpace@MIT",
         collection_handle="123.4/5678",
-        metadata_s3_uri="s3://awd/10.1002-term.3131.json",
+        metadata_s3_uri="s3://dsc/10.1002-term.3131.json",
         bitstream_file_name="10.1002-term.3131.pdf",
-        bitstream_s3_uri="s3://awd/10.1002-term.3131.pdf",
+        bitstream_s3_uri="s3://dsc/10.1002-term.3131.pdf",
     )
     assert dss_message_body == submission_message_body