In 1753 align create statuses across workflows

dsc/cli.py

@@ -133,13 +133,20 @@ def create(
             )
             ctx.exit(exception.exit_code)
 
+    # init list of batch creation errors
+    batch_creation_errors: list = []
     try:
         workflow.create_batch(synced=sync_data)
-    except BatchCreationFailedError:
+    except BatchCreationFailedError as exception:
         logger.error("Failed to create batch")  # noqa: TRY400
+        batch_creation_errors.extend(exception.errors)
 
     if email_recipients:
-        workflow.send_report(step="create", email_recipients=email_recipients.split(","))
+        workflow.send_report(
+            step="create",
+            email_recipients=email_recipients.split(","),
+            errors=batch_creation_errors,
+        )
 
 
 # data sync command

dsc/db/models.py

@@ -24,7 +24,6 @@ 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"
     INGEST_SUCCESS = "ingest_success"

dsc/reports/base.py

@@ -40,9 +40,10 @@ class Report(ABC):
         ),
     )
 
-    def __init__(self, workflow_name: str, batch_id: str):
+    def __init__(self, workflow_name: str, batch_id: str, errors: list | None = None):
         self.workflow_name = workflow_name
         self.batch_id = batch_id
+        self.errors = errors
         self.report_date = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%SZ")
 
         # configure environment for loading jinja templates
@@ -55,8 +56,10 @@ def __init__(self, workflow_name: str, batch_id: str):
         self._item_submissions: list[ItemSubmission] | None = None
 
     @classmethod
-    def load(cls, workflow_name: str, batch_id: str) -> "Report":
-        return cls(workflow_name=workflow_name, batch_id=batch_id)
+    def load(
+        cls, workflow_name: str, batch_id: str, errors: list | None = None
+    ) -> "Report":
+        return cls(workflow_name=workflow_name, batch_id=batch_id, errors=errors)
 
     @property
     @abstractmethod
@@ -97,13 +100,15 @@ def prepare_attachments(self) -> list[tuple]:
         run it. This method returns a list of tuples where each tuple
         contains the filename and a StringIO object (in-memory buffer).
         """
-        return [
-            (
-                f"{self.batch_id}-{attachment.filename}",
-                getattr(self, attachment.method_name)(),
-            )
-            for attachment in self.attachments
-        ]
+        attachments_list: list = []
+        for attachment in self.attachments:
+            content = getattr(self, attachment.method_name)()
+            if content:
+                attachments_list.append(
+                    (f"{self.batch_id}-{attachment.filename}", content)
+                )
+
+        return attachments_list
 
     def upload_attachments(self, output_location: str) -> None:
         for filename, buffer in self.prepare_attachments():
@@ -116,7 +121,9 @@ def upload_attachments(self, output_location: str) -> None:
     # ====================
     # Attachment methods
     # ====================
-    def create_item_submissions_csv(self, fields: list[str] | None = None) -> StringIO:
+    def create_item_submissions_csv(
+        self, fields: list[str] | None = None
+    ) -> StringIO | None:
         """Create a CSV from records in the DynamoDB table for a batch.
 
         This CSV is included in every report that is sent out to provide
@@ -138,9 +145,11 @@ def create_item_submissions_csv(self, fields: list[str] | None = None) -> String
             item_submission.asdict(attrs=fields)
             for item_submission in self.get_item_submissions()
         ]
-        pd.DataFrame(item_submission_dicts).to_csv(buffer, index=False)
-        buffer.seek(0)
-        return buffer
+        if item_submission_dicts:
+            pd.DataFrame(item_submission_dicts).to_csv(buffer, index=False)
+            buffer.seek(0)
+            return buffer
+        return None
 
 
 # =========================
@@ -149,6 +158,11 @@ def create_item_submissions_csv(self, fields: list[str] | None = None) -> String
 
 
 class CreateReport(Report):
+    attachments = (
+        *Report.attachments,
+        Attachment(filename="errors.csv", method_name="create_errors_csv"),
+    )
+
     @property
     def subject(self) -> str:
         return f"DSC Create Batch Results - {self.workflow_name}, batch='{self.batch_id}'"
@@ -165,6 +179,19 @@ def generate_summary(self) -> str:
             item_submissions=self.get_item_submissions(),
         )
 
+    # ====================
+    # Attachment methods
+    # ====================
+    def create_errors_csv(self) -> StringIO | None:
+        if self.errors:
+            buffer = StringIO()
+            pd.DataFrame(self.errors, columns=["item_identifier", "error"]).to_csv(
+                buffer, index=False
+            )
+            buffer.seek(0)
+            return buffer
+        return None
+
 
 class SubmitReport(Report):
     @property

dsc/workflows/base/workflow.py

@@ -254,8 +254,6 @@ def _create_batch_in_db(self, item_submissions: list[ItemSubmission]) -> None:
         """
         for item_submission in item_submissions:
             item_submission.last_run_date = self.run_date
-            item_submission.status = ItemSubmissionStatus.BATCH_CREATED
-            item_submission.status_details = None
             item_submission.save()
 
     def submit_items(self, collection_handle: str | None = None) -> list:
@@ -454,20 +452,27 @@ def workflow_specific_processing(self) -> None:
         )
 
     def send_report(
-        self, step: Literal["create", "submit", "finalize"], email_recipients: list[str]
+        self,
+        step: Literal["create", "submit", "finalize"],
+        email_recipients: list[str],
+        errors: list | None = None,
     ) -> None:
         """Send report as an email via SES.
 
         Args:
             step: The name of the DSC workflow command that is
-                performed. Must be one of ["create", "submit", "finalize"].
+            performed. Must be one of ["create", "submit", "finalize"].
             email_recipients: List of recipient email addresses.
+            errors: This is only used for capturing errors during
+            batch creation.
         """
         logger.info(f"Building report for recipients: {email_recipients}")
 
         # get reporting module for step
         report = self.reporting_modules[step].load(
-            workflow_name=self.workflow_name, batch_id=self.batch_id
+            workflow_name=self.workflow_name,
+            batch_id=self.batch_id,
+            errors=errors,
         )
 
         # upload attachments to S3

dsc/workflows/opencourseware/workflow.py

@@ -6,6 +6,7 @@
 
 import smart_open
 
+from dsc.db.models import ItemSubmissionStatus
 from dsc.exceptions import ItemMetadataNotFoundError
 from dsc.item_submission import ItemSubmission
 from dsc.utils.aws.s3 import S3Client
@@ -125,6 +126,7 @@ def prepare_batch(
                     batch_id=self.batch_id,
                     item_identifier=item_metadata["item_identifier"],
                     workflow_name=self.workflow_name,
+                    status=ItemSubmissionStatus.CREATE_SUCCESS,
                 )
             )
 

dsc/workflows/simple_csv/workflow.py

@@ -5,6 +5,7 @@
 import pandas as pd
 import smart_open
 
+from dsc.db.models import ItemSubmissionStatus
 from dsc.exceptions import ItemBitstreamsNotFoundError
 from dsc.item_submission import ItemSubmission
 from dsc.utils.aws import S3Client
@@ -119,6 +120,7 @@ def prepare_batch(
                     batch_id=self.batch_id,
                     item_identifier=item_metadata["item_identifier"],
                     workflow_name=self.workflow_name,
+                    status=ItemSubmissionStatus.CREATE_SUCCESS,
                 )
             )
 

tests/conftest.py

@@ -62,11 +62,13 @@ def prepare_batch(self, *, synced: bool = False):  # noqa: ARG002
                     batch_id="batch-aaa",
                     item_identifier="123",
                     workflow_name="test",
+                    status=ItemSubmissionStatus.CREATE_SUCCESS,
                 ),
                 ItemSubmission(
                     batch_id="batch-aaa",
                     item_identifier="789",
                     workflow_name="test",
+                    status=ItemSubmissionStatus.CREATE_SUCCESS,
                 ),
             ],
             [],
@@ -211,7 +213,7 @@ def metadata_mapping():
 
 
 @pytest.fixture
-def mocked_item_submission_db(config_instance):
+def mock_item_submission_db(config_instance):
     with mock_aws():
         if not ItemSubmissionDB.exists():
             ItemSubmissionDB.set_table_name(config_instance.item_submissions_table_name)
@@ -220,19 +222,19 @@ def mocked_item_submission_db(config_instance):
 
 
 @pytest.fixture
-def mock_item_submission_db_with_records(mocked_item_submission_db):
+def mock_item_submission_db_with_records(mock_item_submission_db):
     # create two records for 'batch-aaa'
     ItemSubmissionDB(
         batch_id="aaa",
         item_identifier="123",
         workflow_name="test",
-        status=ItemSubmissionStatus.BATCH_CREATED,
+        status=ItemSubmissionStatus.CREATE_SUCCESS,
     ).create()
     ItemSubmissionDB(
         batch_id="aaa",
         item_identifier="456",
         workflow_name="test",
-        status=ItemSubmissionStatus.BATCH_CREATED,
+        status=ItemSubmissionStatus.CREATE_SUCCESS,
     ).create()
 
 

tests/test_cli.py

@@ -13,7 +13,7 @@ def test_create_success(
     caplog,
     runner,
     base_workflow_instance,
-    mocked_item_submission_db,
+    mock_item_submission_db,
     mocked_s3,
     s3_client,
 ):
@@ -43,7 +43,7 @@ def test_create_with_sync_data_success(
     caplog,
     runner,
     base_workflow_instance,
-    mocked_item_submission_db,
+    mock_item_submission_db,
     mocked_s3,
     s3_client,
 ):
@@ -78,7 +78,7 @@ def test_create_with_sync_data_error(
     caplog,
     runner,
     base_workflow_instance,
-    mocked_item_submission_db,
+    mock_item_submission_db,
     mocked_s3,
     s3_client,
 ):
@@ -117,7 +117,7 @@ def test_submit_success(
     mocked_s3,
     mocked_ses,
     mocked_sqs_input,
-    mocked_item_submission_db,
+    mock_item_submission_db,
     base_workflow_instance,
     s3_client,
 ):
@@ -129,13 +129,13 @@ def test_submit_success(
         item_identifier="123",
         batch_id="batch-aaa",
         workflow_name="test",
-        status=ItemSubmissionStatus.BATCH_CREATED,
+        status=ItemSubmissionStatus.CREATE_SUCCESS,
     ).create()
     ItemSubmissionDB(
         item_identifier="789",
         batch_id="batch-aaa",
         workflow_name="test",
-        status=ItemSubmissionStatus.BATCH_CREATED,
+        status=ItemSubmissionStatus.CREATE_SUCCESS,
     ).create()
 
     expected_submission_summary = {"total": 2, "submitted": 2, "skipped": 0, "errors": 0}
@@ -182,7 +182,7 @@ def test_submit_without_collection_handle_success(
     mocked_s3,
     mocked_ses,
     mocked_sqs_input,
-    mocked_item_submission_db,
+    mock_item_submission_db,
     base_workflow_instance,
     s3_client,
 ):
@@ -194,13 +194,13 @@ def test_submit_without_collection_handle_success(
         item_identifier="123",
         batch_id="batch-aaa",
         workflow_name="test",
-        status=ItemSubmissionStatus.BATCH_CREATED,
+        status=ItemSubmissionStatus.CREATE_SUCCESS,
     ).create()
     ItemSubmissionDB(
         item_identifier="789",
         batch_id="batch-aaa",
         workflow_name="test",
-        status=ItemSubmissionStatus.BATCH_CREATED,
+        status=ItemSubmissionStatus.CREATE_SUCCESS,
     ).create()
 
     result = runner.invoke(
@@ -230,7 +230,7 @@ def test_submit_without_collection_handle_success(
 def test_finalize_success(
     caplog,
     runner,
-    mocked_item_submission_db,
+    mock_item_submission_db,
     mocked_ses,
     mocked_sqs_input,
     mocked_sqs_output,

tests/test_db.py

@@ -7,7 +7,7 @@
 from dsc.exceptions import ItemSubmissionCreateError, ItemSubmissionExistsError
 
 
-def test_db_itemsubmission_create_success(mocked_item_submission_db):
+def test_db_itemsubmission_create_success(mock_item_submission_db):
     instance = ItemSubmissionDB(
         batch_id="batch-aaa", item_identifier="123", workflow_name="workflow"
     )
@@ -21,7 +21,7 @@ def test_db_itemsubmission_create_success(mocked_item_submission_db):
     assert fetched_item.workflow_name == "workflow"
 
 
-def test_db_itemsubmission_create_if_exists_raise_error(mocked_item_submission_db):
+def test_db_itemsubmission_create_if_exists_raise_error(mock_item_submission_db):
     instance = ItemSubmissionDB(
         batch_id="batch-aaa", item_identifier="123", workflow_name="workflow"
     )
@@ -35,7 +35,7 @@ def test_db_itemsubmission_create_if_exists_raise_error(mocked_item_submission_d
 
 @patch("dsc.db.models.ItemSubmissionDB.save")
 def test_db_itemsubmission_create_if_puterror_raise_error(
-    mock_item_submission_db_save, mocked_item_submission_db
+    mock_item_submission_db_save, mock_item_submission_db
 ):
 
     class MockDynamoDBError(Exception):