Implement batch preparation methods for digitized theses workflow

Why these changes are being introduced:
* The batch preparation methods for the digitized theses workflow
must be capable of performing the following functions:

1. Sync the batch folder from the digitized theses workflow S3 bucket
to a minted batch folder in the DSC S3 bucket
2. Download metadata from Alma via SRU
3. Get an item from DSpace
4. Determine if an item is a valid 'replacement thesis', given DSpace item metadata
5. Organize the contents of the minted batch folder into 'new-theses/' and
'replacement-theses/' subfolders

How this addresses that need:
* Implement methods on DigitizedThesesWorkflow for batch preparation functions
* Expand ItemSubmissionStatus to `CREATE_SUCCESS`, `CREATE_FAILED`, `CREATE_SKIPPED`
* Add exceptions to handle item retrieval from DSpace

Side effects of this change:
* Though we've expanded ItemSubmissionStatus to include `CREATE_*` statuses,
`BATCH_CREATED` remains to support backwards compatibility with other workflows.
Updating all workflows to use `CREATE_*` will be in the DSC backlog for future
work.

Relevant ticket(s):
* https://mitlibraries.atlassian.net/browse/IN-1626

Author: jonavellecuerdo

dsc/config.py

@@ -83,11 +83,13 @@ def warning_only_loggers(self) -> list:
 
     # Workflow-specific env vars
     @property
-    def dspace_credentials(self) -> str:
+    def dspace_credentials(self) -> dict:
         value = os.getenv("DSPACE_CREDENTIALS")
         if not value:
             raise OSError("Env var 'DSPACE_CREDENTIALS' must be defined")
-        return value
+        credentials = json.loads(value)
+
+        return {"IR-8": credentials["ir-8"], "DDC-8": credentials["ddc-8"]}
 
     @property
     def metadata_api_url(self) -> str:

dsc/db/models.py

@@ -21,6 +21,9 @@
 
 
 class ItemSubmissionStatus(StrEnum):
+    CREATE_SUCCESS = "create_success"
+    CREATE_FAILED = "create_failed"
+    CREATE_SKIPPED = "create_skipped"
     BATCH_CREATED = "batch_created"
     SUBMIT_SUCCESS = "submit_success"
     SUBMIT_FAILED = "submit_failed"

dsc/exceptions.py

@@ -30,6 +30,43 @@ class SQSMessageSendError(Exception):
     pass
 
 
+# Exceptions for DSpace client
+class DSpaceClientError(Exception):
+    """General exception raised when DSpace client action results in error."""
+
+
+class DSpaceClientCredentialsNotFoundError(DSpaceClientError):
+    """Raise when env var DSPACE_CREDENTIALS does not include submission system."""
+
+
+class DSpaceClientAuthenticationError(DSpaceClientError):
+    """Raise when DSpace client fails authentication.
+
+    Authentication may fail due to 401 Unauthorized or 403 Forbidden errors.
+    """
+
+    def __init__(
+        self,
+        dspace_url: str | float | None,
+        dspace_user: str | float | None,
+    ):
+        self.message = (
+            f"Failed to authenticate to DSpace server at '{dspace_url}' with user "
+            f"'{dspace_user}'. Please verify that the DSPACE_CREDENTIALS "
+            "environment variable is set correctly and that the DSpace server is "
+            "accessible."
+        )
+
+
+class DSpaceClientSearchError(DSpaceClientError):
+    """Raise when DSpace client search operation results in error.
+
+    Search is performed by dspace_rest_client.client.search_objects,
+    which returns None if the response from a GET request returns an
+    exit code other than 200.
+    """
+
+
 # Exceptions for 'create-batch' step
 class BatchCreationFailedError(Exception):
     def __init__(self, errors: list[tuple]) -> None:

dsc/workflows/digitized_theses/__init__.py

@@ -1,3 +1,4 @@
-from dsc.workflows.digitized_theses.transformer import DigitizedThesesTransformer, NSMAP
+from dsc.workflows.digitized_theses.transformer import NSMAP, DigitizedThesesTransformer
+from dsc.workflows.digitized_theses.workflow import DigitizedTheses
 
-__all__ = ["DigitizedThesesTransformer", "NSMAP"]
+__all__ = ["NSMAP", "DigitizedTheses", "DigitizedThesesTransformer"]