refactor: tweaks around submission status - now optional to pass as can determine from audit (in service based approach). Tests fixed and added to

Author: stevenhsd

src/dve/core_engine/backends/base/auditing.py

@@ -330,7 +330,7 @@ def mark_business_rules(self, submissions: list[tuple[str, bool]], **kwargs):
             ProcessingStatusRecord(
                 submission_id=submission_id,
                 processing_status="business_rules",
-                submission_result="failed" if failed else None,
+                submission_result="validation_failed" if failed else None,
                 **kwargs,
             )
             for submission_id, failed in submissions
@@ -380,7 +380,10 @@ def mark_failed(self, submissions: list[str], **kwargs):
         """Update submission processing_status to failed."""
         recs = [
             ProcessingStatusRecord(
-                submission_id=submission_id, processing_status="failed", **kwargs
+                submission_id=submission_id,
+                processing_status="failed",
+                submission_result="processing_failed",
+                **kwargs
             )
             for submission_id in submissions
         ]
@@ -494,16 +497,22 @@ def get_submission_statistics(self, submission_id: str) -> Optional[SubmissionSt
             )
         except StopIteration:
             return None
-    def get_submission_status(self, submission_id: str) -> SubmissionStatus:
+    def get_submission_status(self, submission_id: str) -> Optional[SubmissionStatus]:
         """Get the latest submission status for a submission"""
+        
+        try:
+            processing_rec: ProcessingStatusRecord = next(self._processing_status.conv_to_records(
+                self._processing_status.get_most_recent_records(
+                    order_criteria=[OrderCriteria("time_updated", True)],
+                    pre_filter_criteria=[FilterCriteria("submission_id", submission_id)]
+                    )))
+        except StopIteration:
+            return None
         sub_status = SubmissionStatus()
-        processing_rec: ProcessingStatusRecord = next(self._processing_status.conv_to_records(self._processing_status.get_most_recent_records(order_criteria=[OrderCriteria("time_updated", True)],
-                                                        pre_filter_criteria=[FilterCriteria("submission_id",
-                                                                                            submission_id)])))
         sub_stats_rec: Optional[SubmissionStatisticsRecord] = self.get_submission_statistics(submission_id)
-        if processing_rec.processing_status == "failed":
+        if processing_rec.submission_result == "processing_failed":
             sub_status.processing_failed = True
-        if processing_rec.submission_result == "failed":
+        if processing_rec.submission_result == "validation_failed":
             sub_status.validation_failed = True
         if sub_stats_rec:
             sub_status.number_of_records = sub_stats_rec.record_count

src/dve/core_engine/type_hints.py

@@ -236,7 +236,7 @@
 PROCESSING_STATUSES: tuple[ProcessingStatus, ...] = tuple(list(get_args(ProcessingStatus)))
 """List of all possible DVE submission statuses"""
 
-SubmissionResult = Literal["success", "failed", "archived", "processing_failed"]
+SubmissionResult = Literal["success", "validation_failed", "archived", "processing_failed"]
 """Allowed DVE submission results"""
 
 SUBMISSION_RESULTS: tuple[SubmissionResult, ...] = tuple(list(get_args(SubmissionResult)))

src/dve/pipeline/foundry_ddb_pipeline.py

@@ -41,6 +41,7 @@ def file_transformation(
                     "file_transformation",
                     [CriticalProcessingError.from_exception(exc)]
                 )
+            self._audit_tables.mark_failed(submissions=[submission_info.submission_id])
             return submission_info, SubmissionStatus(processing_failed=True)
 
     def apply_data_contract(self, submission_info: SubmissionInfo, submission_status: SubmissionStatus) -> tuple[SubmissionInfo | SubmissionStatus]:
@@ -54,6 +55,7 @@ def apply_data_contract(self, submission_info: SubmissionInfo, submission_status
                     "contract",
                     [CriticalProcessingError.from_exception(exc)]
                 )
+            self._audit_tables.mark_failed(submissions=[submission_info.submission_id])
             return submission_info, SubmissionStatus(processing_failed=True)
 
     def apply_business_rules(self, submission_info: SubmissionInfo, submission_status: SubmissionStatus):
@@ -67,6 +69,7 @@ def apply_business_rules(self, submission_info: SubmissionInfo, submission_statu
                     "business_rules",
                     [CriticalProcessingError.from_exception(exc)]
                 )
+            self._audit_tables.mark_failed(submissions=[submission_info.submission_id])
             return submission_info, SubmissionStatus(processing_failed=True)
 
     def error_report(self, submission_info: SubmissionInfo, submission_status: SubmissionStatus):
@@ -82,12 +85,14 @@ def error_report(self, submission_info: SubmissionInfo, submission_status: Submi
                     "error_report",
                     [CriticalProcessingError.from_exception(exc)]
                 )
+            self._audit_tables.mark_failed(submissions=[submission_info.submission_id])
             return submission_info, submission_status, sub_stats, report_uri
 
-    def run_pipeline(self, submission_info: SubmissionInfo) -> tuple[Optional[URI], URI, URI]:
+    def run_pipeline(self, submission_info: SubmissionInfo) -> tuple[Optional[URI], Optional[URI], URI]:
         """Sequential single submission pipeline runner"""
         try:
             sub_id: str = submission_info.submission_id
+            report_uri = None
             self._audit_tables.add_new_submissions(submissions=[submission_info])
             self._audit_tables.mark_transform(submission_ids=[sub_id])
             sub_info, sub_status = self.file_transformation(submission_info=submission_info)
@@ -99,13 +104,14 @@ def run_pipeline(self, submission_info: SubmissionInfo) -> tuple[Optional[URI],
                     submission_info=submission_info, submission_status=sub_status
                 )
 
-            self._audit_tables.mark_error_report(
-                submissions=[(sub_id, sub_status.submission_result)]
-            )
-            sub_info, sub_status, sub_stats, report_uri = self.error_report(
-                submission_info=submission_info, status=sub_status
-            )
-            self._audit_tables.add_submission_statistics_records(sub_stats=[sub_stats])
+            if not sub_status.processing_failed:
+                self._audit_tables.mark_error_report(
+                    submissions=[(sub_id, sub_status.submission_result)]
+                )
+                sub_info, sub_status, sub_stats, report_uri = self.error_report(
+                    submission_info=submission_info, submission_status=sub_status
+                )
+                self._audit_tables.add_submission_statistics_records(sub_stats=[sub_stats])
         except Exception as err: # pylint: disable=W0718
             self._logger.error(
                 f"During processing of submission_id: {sub_id}, the following exception was raised: {err}"
@@ -124,6 +130,6 @@ def run_pipeline(self, submission_info: SubmissionInfo) -> tuple[Optional[URI],
                 if (sub_status.validation_failed or sub_status.processing_failed)
                 else fh.joinuri(self.processed_files_path, sub_id, "business_rules")
             ),
-            report_uri,
+            report_uri if report_uri else None,
             audit_files_uri,
         )

src/dve/pipeline/pipeline.py

@@ -362,7 +362,7 @@ def file_transformation_step(
                     starmap(
                         lambda x, _: (
                             x.submission_id,
-                            "failed",
+                            "validation_failed",
                         ),
                         failed,
                     )
@@ -377,9 +377,10 @@ def file_transformation_step(
 
         return success, failed
 
-    def apply_data_contract(self, submission_info: SubmissionInfo, submission_status: SubmissionStatus) -> tuple[SubmissionInfo, SubmissionStatus]:
+    def apply_data_contract(self, submission_info: SubmissionInfo, submission_status: Optional[SubmissionStatus] = None) -> tuple[SubmissionInfo, SubmissionStatus]:
         """Method for applying the data contract given a submission_info"""
-
+        if not submission_status:
+            submission_status = self._audit_tables.get_submission_status(submission_info.submission_id) 
         if not self.processed_files_path:
             raise AttributeError("processed files path not provided")
 
@@ -422,14 +423,17 @@ def apply_data_contract(self, submission_info: SubmissionInfo, submission_status
         return submission_info, submission_status
 
     def data_contract_step(
-        self, pool: Executor, file_transform_results: list[tuple[SubmissionInfo, SubmissionStatus]]
+        self, pool: Executor, file_transform_results: list[tuple[SubmissionInfo, Optional[SubmissionStatus]]]
     ) -> tuple[list[tuple[SubmissionInfo, SubmissionStatus]], list[tuple[SubmissionInfo, SubmissionStatus]]]:
         """Step to validate the types of an untyped (stringly typed) parquet file"""
         processed_files: list[tuple[SubmissionInfo, SubmissionStatus]] = []
         failed_processing: list[tuple[SubmissionInfo, SubmissionStatus]] = []
         dc_futures: list[tuple[SubmissionInfo, SubmissionStatus, Future]] = []
 
         for info, sub_status in file_transform_results:
+            sub_status = (
+                sub_status if sub_status 
+                else self._audit_tables.get_submission_status(info.submission_id))
             dc_futures.append((info, sub_status, pool.submit(self.apply_data_contract, info, sub_status)))
 
         for sub_info, sub_status, future in dc_futures:
@@ -476,10 +480,13 @@ def data_contract_step(
 
         return processed_files, failed_processing
 
-    def apply_business_rules(self, submission_info: SubmissionInfo, submission_status: SubmissionStatus):
+    def apply_business_rules(self, submission_info: SubmissionInfo, submission_status: Optional[SubmissionStatus] = None):
         """Apply the business rules to a given submission, the submission may have failed at the
         data_contract step so this should be passed in as a bool
         """
+        if not submission_status:
+            submission_status = self._audit_tables.get_submission_status(submission_info.submission_id)
+
         if not self.rules_path:
             raise AttributeError("business rules path not provided.")
 
@@ -557,7 +564,7 @@ def apply_business_rules(self, submission_info: SubmissionInfo, submission_statu
     def business_rule_step(
         self,
         pool: Executor,
-        files: list[tuple[SubmissionInfo, SubmissionStatus]],
+        files: list[tuple[SubmissionInfo, Optional[SubmissionStatus]]],
     ) -> tuple[
         list[tuple[SubmissionInfo, SubmissionStatus]],
         list[tuple[SubmissionInfo, SubmissionStatus]],
@@ -567,6 +574,9 @@ def business_rule_step(
         future_files: list[tuple[SubmissionInfo, SubmissionStatus, Future]] = []
 
         for submission_info, submission_status in files:
+            submission_status = (
+                submission_status if submission_status 
+                else self._audit_tables.get_submission_status(submission_info.submission_id))
             future_files.append(
                 (
                     submission_info,
@@ -677,8 +687,14 @@ def _get_error_dataframes(self, submission_id: str):
 
         return errors_df, aggregates
 
-    def error_report(self, submission_info: SubmissionInfo, submission_status: SubmissionStatus) -> tuple[SubmissionInfo, SubmissionStatus, Optional[SubmissionStatisticsRecord], Optional[URI]]:
+    def error_report(self,
+                     submission_info: SubmissionInfo,
+                     submission_status: Optional[SubmissionStatus] = None) -> tuple[SubmissionInfo, SubmissionStatus, Optional[SubmissionStatisticsRecord], Optional[URI]]:
         """Creates the error reports given a submission info and submission status"""
+        
+        if not submission_status:
+            submission_status = self._audit_tables.get_submission_status(submission_info.submission_id)
+
         if not self.processed_files_path:
             raise AttributeError("processed files path not provided")
 
@@ -728,7 +744,7 @@ def error_report(self, submission_info: SubmissionInfo, submission_status: Submi
     def error_report_step(
         self,
         pool: Executor,
-        processed: Iterable[tuple[SubmissionInfo, SubmissionStatus]] = tuple(),
+        processed: Iterable[tuple[SubmissionInfo, Optional[SubmissionStatus]]] = tuple(),
         failed_file_transformation: Iterable[tuple[SubmissionInfo, SubmissionStatus]] = tuple(),
     ) -> list[
         tuple[SubmissionInfo, SubmissionStatus, Union[None, SubmissionStatisticsRecord], URI]
@@ -743,6 +759,10 @@ def error_report_step(
         failed_processing: list[tuple[SubmissionInfo, SubmissionStatus]] = []
 
         for info, status in processed:
+            status = (
+                status if status
+                else self._audit_tables.get_submission_status(info.submission_id)
+                )
             futures.append((info, status, pool.submit(self.error_report, info, status)))
 
         for info_dict, status in failed_file_transformation:

src/dve/pipeline/utils.py

@@ -86,5 +86,5 @@ def submission_result(self) -> SubmissionResult:
         if self.processing_failed:
             return "processing_failed"
         if self.validation_failed:
-            return "failed"
+            return "validation_failed"
         return "success"

src/dve/reporting/utils.py

@@ -48,16 +48,16 @@ def dump_processing_errors(
     if not working_folder:
         raise AttributeError("processed files path not passed")
 
-    errors = fh.joinuri(working_folder, "errors", f"processing_errors.json")
+    error_file: URI = fh.joinuri(working_folder, "errors", f"processing_errors.json")
     processed = []
 
     for error in errors:
         processed.append({"step_name": step_name,
                           "error_location": "processing",
                           "error_level": "integrity",
-                          "error_message": repr(error)})
+                          "error_message": error.error_message})
 
-    with fh.open_stream(errors, "a") as f:
+    with fh.open_stream(error_file, "a") as f:
         json.dump(
             processed,
             f,

tests/features/planets.feature

@@ -42,9 +42,7 @@ Feature: Pipeline tests using the planets dataset
         And I add initial audit entries for the submission
         Then the latest audit record for the submission is marked with processing status file_transformation
         When I run the file transformation phase
-        Then the latest audit record for the submission is marked with processing status error_report
-        When I run the error report phase
-        Then An error report is produced
+        Then the latest audit record for the submission is marked with processing status failed
 
     Scenario: Handle a file with duplicated extension provided (spark)
         Given I submit the planets file planets.csv.csv for processing