Add batch creation methods to the Workflow abstract base class

Why these changes are being introduced:
* Support workflows that require programmatic batch creation.
It was decided that the batch creation step would assume
the responsibilities that were previously assigned to the
reconcile step, which is now up for deprecation.

How this addresses that need:
* Define shared private method for writing records to DynamoDB
* Add abstract method custom batch preparation processes
* Define SimpleCSV.prepare_batch
* Add unit tests for batch creation methods
* Remove record creation from Workflow.reconcile_items

Side effects of this change:
* Sets the stage for deprecation of 'reconcile_items'

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

Author: jonavellecuerdo

dsc/db/models.py

@@ -21,6 +21,7 @@
 
 
 class ItemSubmissionStatus(StrEnum):
+    BATCH_CREATED = "batch_created"
     RECONCILE_SUCCESS = "reconcile_success"
     RECONCILE_FAILED = "reconcile_failed"
     SUBMIT_SUCCESS = "submit_success"

dsc/exceptions.py

@@ -69,3 +69,18 @@ def __str__(self) -> str:
         if len(self.metadata_without_bitstreams) > 20:  # noqa: PLR2004
             message = f"{message} (showing first 20 metadata item identifiers)"
         return f"{message}: {self.metadata_without_bitstreams}"
+
+
+# Exceptions for 'create-batch' step
+class BatchCreationFailedError(Exception):
+    pass
+
+
+class ItemMetadataNotFoundError(Exception):
+    def __init__(self) -> None:
+        super().__init__("No metadata found for the item submission")
+
+
+class ItemBitstreamsNotFoundError(Exception):
+    def __init__(self) -> None:
+        super().__init__("No bitstreams found for the item submission")

dsc/workflows/base/__init__.py

@@ -14,6 +14,7 @@
 from dsc.config import Config
 from dsc.db.models import ItemSubmissionDB, ItemSubmissionStatus
 from dsc.exceptions import (
+    BatchCreationFailedError,
     InvalidSQSMessageError,
     InvalidWorkflowNameError,
     ReconcileFailedError,
@@ -221,6 +222,60 @@ def item_metadata_iter(self) -> Iterator[dict[str, Any]]:
         MUST be overridden by workflow subclasses.
         """
 
+    @final
+    def create_batch(self) -> None:
+        """Create a batch of item submissions for processing.
+
+        A "batch" refers to a collection of item submissions that are grouped together
+        for coordinated processing, storage, and workflow execution. Each batch
+        typically consists of multiple items, each with its own metadata and
+        associated files, organized under a unique batch identifier.
+
+        This method prepares the necessary assets in S3 (programmatically as needed)
+        and records each item in the batch to DynamoDB.
+        """
+        item_submissions, errors = self.prepare_batch()
+        if errors:
+            raise BatchCreationFailedError
+        self._create_batch_in_db(item_submissions)
+
+    @abstractmethod
+    def prepare_batch(self) -> tuple[list, ...]:
+        """Prepare batch submission assets in S3.
+
+        This method performs the required steps to prepare a batch
+        of item submissions in S3. These steps must include (at minimum)
+        the following checks:
+
+        - Check if there is metadata for the item submission;
+          otherwise raise dsc.exceptions.ItemMetadataNotFoundError
+        - Check if there are any bitstreams for the item submission;
+          otherwise raise dsc.exceptions.ItemBitstreamsNotFoundError
+
+        MUST be overridden by workflow subclasses.
+
+        Returns:
+            A tuple of item submissions (init params) represented as a
+            list of dicts and errors represented as a list of tuples
+            containing the item identifier and the error message.
+        """
+        pass  # noqa: PIE790
+
+    @final
+    def _create_batch_in_db(self, item_submissions: list[dict]) -> None:
+        """Write records for a batch of item submissions to DynamoDB.
+
+        This method loops through the item submissions (init params)
+        represented as a list dicts. For each item submission, the
+        method creates an instance of ItemSubmission and saves the
+        record to DynamoDB.
+        """
+        for item_submission_init_params in item_submissions:
+            item_submission = ItemSubmission.create(**item_submission_init_params)
+            item_submission.last_run_date = self.run_date
+            item_submission.status = ItemSubmissionStatus.BATCH_CREATED
+            item_submission.save()
+
     @final
     def reconcile_items(self) -> bool:
         """Reconcile item submissions for a batch.
@@ -241,20 +296,24 @@ def reconcile_items(self) -> bool:
         NOTE: This method is likely the first time a record will be inserted
         into DynamoDB for each item submission. If already present,
         its status will be updated.
+
+        TODO: Reconcile methods will be deprecated after end-to-end testing.
         """
         reconciled_items = {}  # key=item_identifier, value=list of bitstream URIs
         bitstreams_without_metadata = []  # list of bitstream URIs
         metadata_without_bitstreams = []  # list of item identifiers
 
         # loop through each item metadata
         for item_metadata in self.item_metadata_iter():
-            item_submission = ItemSubmission.get_or_create(
+            item_submission = ItemSubmission.get(
                 batch_id=self.batch_id,
                 item_identifier=item_metadata["item_identifier"],
-                workflow_name=self.workflow_name,
-                source_system_identifier=item_metadata.get("source_system_identifier"),
             )
 
+            # if no corresponding record in DynamoDB, skip
+            if not item_submission:
+                continue
+
             # attach source metadata
             item_submission.source_metadata = item_metadata
 
@@ -336,8 +395,7 @@ def reconcile_items(self) -> bool:
         )
         return True
 
-    @abstractmethod
-    def reconcile_item(self, item_submission: ItemSubmission) -> bool:
+    def reconcile_item(self, _item_submission: ItemSubmission) -> bool:
         """Reconcile bitstreams and metadata for an item.
 
         Items in DSpace represent a "work" and combine metadata and files,
@@ -348,7 +406,10 @@ def reconcile_item(self, item_submission: ItemSubmission) -> bool:
 
         If an item fails reconcile, this method should raise
         dsc.exceptions.ReconcileFailed*Error. Otherwise, return True.
+
+        TODO: Reconcile methods will be deprecated after end-to-end testing.
         """
+        return False
 
     def _report_reconcile_workflow_events(
         self,

dsc/workflows/base/simple_csv.py

@@ -5,7 +5,10 @@
 import pandas as pd
 import smart_open
 
-from dsc.exceptions import ReconcileFailedMissingBitstreamsError
+from dsc.exceptions import (
+    ItemBitstreamsNotFoundError,
+    ReconcileFailedMissingBitstreamsError,
+)
 from dsc.item_submission import ItemSubmission
 from dsc.utilities.aws import S3Client
 from dsc.workflows.base import Workflow
@@ -74,7 +77,6 @@ def item_metadata_iter(
                 logger.warning(
                     f"Renaming multiple columns as 'item_identifier': {col_names}"
                 )
-
             metadata_df = metadata_df.rename(
                 columns={
                     col: "item_identifier"
@@ -92,3 +94,31 @@ def item_metadata_iter(
 
             for _, row in metadata_df.iterrows():
                 yield row.to_dict()
+
+    def prepare_batch(self) -> tuple[list, ...]:
+        item_submissions = []
+        errors = []
+
+        for item_metadata in self.item_metadata_iter():
+            # check if there are any bitstreams associated with the item submission
+            if not self.get_item_bitstream_uris(
+                item_identifier=item_metadata["item_identifier"]
+            ):
+                errors.append(
+                    (
+                        item_metadata["item_identifier"],
+                        str(ItemBitstreamsNotFoundError()),
+                    )
+                )
+                continue
+
+            # if item submission has associated bitstreams
+            # save init params
+            item_submissions.append(
+                {
+                    "batch_id": self.batch_id,
+                    "item_identifier": item_metadata["item_identifier"],
+                    "workflow_name": self.workflow_name,
+                }
+            )
+        return item_submissions, errors

tests/conftest.py

@@ -24,6 +24,7 @@
 from dsc.workflows.base.simple_csv import SimpleCSV
 
 
+# Test Workflow classes ######################
 class TestWorkflow(Workflow):
 
     workflow_name: str = "test"
@@ -61,6 +62,47 @@ def item_metadata_iter(self):
             },
         ]
 
+    def prepare_batch(self):
+        return (
+            [
+                {
+                    "batch_id": "batch-aaa",
+                    "item_identifier": "123",
+                    "workflow_name": "test",
+                },
+                {
+                    "batch_id": "batch-aaa",
+                    "item_identifier": "789",
+                    "workflow_name": "test",
+                },
+            ],
+            [],
+        )
+
+
+class TestOpenCourseWare(OpenCourseWare):
+
+    @property
+    def output_queue(self) -> str:
+        return "mock-output-queue"
+
+    def prepare_batch(self):
+        return (
+            [
+                {
+                    "batch_id": "batch-aaa",
+                    "item_identifier": "123",
+                    "workflow_name": "opencourseware",
+                },
+                {
+                    "batch_id": "batch-aaa",
+                    "item_identifier": "124",
+                    "workflow_name": "opencourseware",
+                },
+            ],
+            [],
+        )
+
 
 class TestSimpleCSV(SimpleCSV):
 
@@ -80,13 +122,30 @@ def output_queue(self) -> str:
         return "mock-output-queue"
 
 
-class TestOpenCourseWare(OpenCourseWare):
+# Test Workflow instances ####################
+@pytest.fixture
+@freeze_time("2025-01-01 09:00:00")
+def base_workflow_instance(item_metadata, metadata_mapping, mocked_s3):
+    return TestWorkflow(batch_id="batch-aaa")
 
-    @property
-    def output_queue(self) -> str:
-        return "mock-output-queue"
+
+@pytest.fixture
+def simple_csv_workflow_instance(metadata_mapping):
+    return TestSimpleCSV(batch_id="batch-aaa")
+
+
+@pytest.fixture
+@freeze_time("2025-01-01 09:00:00")
+def archivesspace_workflow_instance(tmp_path):
+    return ArchivesSpace(batch_id="batch-aaa")
+
+
+@pytest.fixture
+def opencourseware_workflow_instance():
+    return TestOpenCourseWare(batch_id="batch-aaa")
 
 
+# Test fixtures ##############################
 @pytest.fixture(autouse=True)
 def _test_env(monkeypatch):
     monkeypatch.setenv("SENTRY_DSN", "None")
@@ -113,28 +172,6 @@ def moto_server():
     time.sleep(0.5)
 
 
-@pytest.fixture
-@freeze_time("2025-01-01 09:00:00")
-def base_workflow_instance(item_metadata, metadata_mapping, mocked_s3):
-    return TestWorkflow(batch_id="batch-aaa")
-
-
-@pytest.fixture
-def simple_csv_workflow_instance(metadata_mapping):
-    return TestSimpleCSV(batch_id="batch-aaa")
-
-
-@pytest.fixture
-@freeze_time("2025-01-01 09:00:00")
-def archivesspace_workflow_instance(tmp_path):
-    return ArchivesSpace(batch_id="batch-aaa")
-
-
-@pytest.fixture
-def opencourseware_workflow_instance():
-    return TestOpenCourseWare(batch_id="batch-aaa")
-
-
 @pytest.fixture
 def config_instance():
     return Config()

tests/test_cli.py

@@ -22,6 +22,11 @@ def test_reconcile_success(
     s3_client.put_file(
         file_content="", bucket="dsc", key="simple-csv/batch-aaa/123_002.jpg"
     )
+    instance = ItemSubmissionDB(
+        batch_id="batch-aaa", item_identifier="123", workflow_name="workflow"
+    )
+    instance.create()
+
     result = runner.invoke(
         main,
         [

tests/test_workflow_base.py

@@ -1,8 +1,10 @@
 import json
+from datetime import UTC, datetime
 from unittest.mock import patch
 
 import pytest
 from botocore.exceptions import ClientError
+from freezegun import freeze_time
 
 from dsc.db.models import ItemSubmissionDB, ItemSubmissionStatus
 from dsc.exceptions import (
@@ -38,6 +40,18 @@ def test_base_workflow_get_workflow_invalid_workflow_name_raises_error(
         base_workflow_instance.get_workflow("tast")
 
 
+@freeze_time("2025-01-01 09:00:00")
+def test_base_workflow_create_batch_in_db_success(
+    base_workflow_instance, mocked_item_submission_db
+):
+    item_submissions, _ = base_workflow_instance.prepare_batch()
+    base_workflow_instance._create_batch_in_db(item_submissions)  # noqa: SLF001
+    item_submission = ItemSubmissionDB.get(hash_key="batch-aaa", range_key="123")
+
+    assert item_submission.last_run_date == datetime(2025, 1, 1, 9, 0, tzinfo=UTC)
+    assert item_submission.status == ItemSubmissionStatus.BATCH_CREATED
+
+
 def test_base_workflow_submit_items_success(
     caplog,
     base_workflow_instance,

tests/test_workflow_opencourseware.py

@@ -120,6 +120,12 @@ def test_workflow_ocw_reconcile_items_success(
         bucket="dsc",
         key="opencourseware/batch-aaa/123.zip",
     )
+    ItemSubmissionDB(
+        item_identifier="123",
+        batch_id="batch-aaa",
+        workflow_name="opencourseware",
+    ).create()
+
     mock_opencourseware_read_metadata_from_zip_file.return_value = (
         opencourseware_source_metadata
     )
@@ -157,6 +163,17 @@ def test_workflow_ocw_reconcile_items_if_not_reconciled_success(
         bucket="dsc",
         key="opencourseware/batch-aaa/124.zip",
     )
+    ItemSubmissionDB(
+        item_identifier="123",
+        batch_id="batch-aaa",
+        workflow_name="opencourseware",
+    ).create()
+    ItemSubmissionDB(
+        item_identifier="124",
+        batch_id="batch-aaa",
+        workflow_name="opencourseware",
+    ).create()
+
     mock_opencourseware_read_metadata_from_zip_file.side_effect = [
         opencourseware_source_metadata,
         FileNotFoundError,