refactor: add rejected records to sub status object (#122)

Author: georgeRobertson

src/dve/pipeline/pipeline.py

@@ -42,7 +42,7 @@
 from dve.parser.file_handling.implementations.file import LocalFilesystemImplementation
 from dve.parser.file_handling.service import _get_implementation
 from dve.pipeline.utils import SubmissionStatus, deadletter_file, load_config, load_reader
-from dve.reporting.constants import ErrorReportStatus
+from dve.reporting.constants import ErrorReportCategories
 from dve.reporting.error_report import ERROR_SCHEMA, calculate_aggregates
 
 PERMISSIBLE_EXCEPTIONS: tuple[type[Exception]] = (
@@ -635,14 +635,19 @@ def apply_business_rules(  # pylint: disable=R0914
                 projected
             )
 
-        # todo - add to submission_status around records that have passed record validations/rejected
         submission_status.number_of_records = self.get_entity_count(
             entity=entity_manager.entities[
                 f"""Original{rules.global_variables.get(
                                               'entity',
                                               submission_info.dataset_id)}"""
             ]
         )
+        submission_status.number_of_records_rejected = (
+            submission_status.number_of_records -
+            self.get_entity_count(entity_manager.entities[
+                rules.global_variables.get("entity", submission_info.dataset_id)
+            ])
+        )
 
         return submission_info, submission_status
 
@@ -765,8 +770,8 @@ def _get_error_dataframes(self, submission_id: str):
                     pl.when(pl.col("Status") == pl.lit("informational"))
                     .then(pl.lit("Warning"))
                     .when(pl.col("FailureType") == pl.lit("submission"))  # type: ignore
-                    .then(pl.lit(ErrorReportStatus.FILE_REJECTION.reporting_name))  # type: ignore
-                    .otherwise(pl.lit(ErrorReportStatus.RECORD_REJECTION.reporting_name))  # type: ignore
+                    .then(pl.lit(ErrorReportCategories.FILE_REJECTION.reporting_name))  # type: ignore
+                    .otherwise(pl.lit(ErrorReportCategories.RECORD_REJECTION.reporting_name))  # type: ignore
                     .alias("error_type")  # type: ignore
                 )
                 df = df.select(
@@ -830,12 +835,12 @@ def error_report(
                 submission_id=submission_info.submission_id,
                 record_count=submission_status.number_of_records,
                 number_submission_rejections=err_types.get(
-                    ErrorReportStatus.FILE_REJECTION.reporting_name, 0
+                    ErrorReportCategories.FILE_REJECTION.reporting_name, 0
                 ),
                 number_record_rejections=err_types.get(
-                    ErrorReportStatus.RECORD_REJECTION.reporting_name, 0
+                    ErrorReportCategories.RECORD_REJECTION.reporting_name, 0
                 ),
-                number_warnings=err_types.get(ErrorReportStatus.WARNING.reporting_name, 0),
+                number_warnings=err_types.get(ErrorReportCategories.WARNING.reporting_name, 0),
             )
 
         summary_dict = {
@@ -846,7 +851,7 @@ def error_report(
         summary_items = er.SummaryItems(
             submission_status=submission_status,
             summary_dict=summary_dict,
-            row_headings=[e.reporting_name for e in ErrorReportStatus],
+            row_headings=[e.reporting_name for e in ErrorReportCategories],
         )
 
         workbook = er.ExcelFormat(

src/dve/pipeline/utils.py

@@ -86,10 +86,12 @@ def __init__(
         self,
         validation_failed: bool = False,
         number_of_records: Optional[int] = None,
+        number_of_records_rejected: Optional[int] = None,
         processing_failed: bool = False,
     ):
         self.validation_failed = validation_failed
         self.number_of_records = number_of_records
+        self.number_of_records_rejected = number_of_records_rejected
         self.processing_failed = processing_failed
 
     @property

src/dve/reporting/constants.py

@@ -5,9 +5,9 @@
 from enum import Enum
 
 
-class ErrorReportStatus(Enum):
+class ErrorReportCategories(Enum):
     """
-    Constant to centrally hold error report status.
+    Categories available within the error report.
     """
 
     FILE_REJECTION = 1, "File Rejection"
@@ -20,3 +20,15 @@ def reporting_name(self):
         The error report 'friendly' name.
         """
         return self.value[1]
+
+
+class ErrorReportStatus:  # pylint: disable=R0903
+    """
+    Statuses available within the error report.
+    """
+
+    PROCESSING_FAILED = "There was an issue processing the submission. Please contact support."
+    FILE_REJECTION = "File has been rejected"
+    RECORD_REJECTION = "File has been accepted with record rejections"
+    ACCEPTED_WITH_WARNING = "File has been accepted, all records accepted with warnings"
+    ACCEPTED = "File has been accepted, no issues to report"

src/dve/reporting/error_report.py

@@ -11,7 +11,7 @@
 from dve.common.error_utils import conditional_cast
 from dve.core_engine.message import FeedbackMessage
 from dve.parser.file_handling.service import open_stream
-from dve.reporting.constants import ErrorReportStatus
+from dve.reporting.constants import ErrorReportCategories
 
 ERROR_SCHEMA = {
     "Table": Utf8(),
@@ -86,7 +86,7 @@ def create_error_dataframe(errors: deque[FeedbackMessage], key_fields):
 
     df = df.with_columns(  # type: ignore
         pl.when(pl.col("Status") == pl.lit("error"))  # type: ignore
-        .then(pl.lit(ErrorReportStatus.FILE_REJECTION.reporting_name))  # type: ignore
+        .then(pl.lit(ErrorReportCategories.FILE_REJECTION.reporting_name))  # type: ignore
         .otherwise(pl.lit("Warning"))  # type: ignore
         .alias("error_type")
     )

src/dve/reporting/excel_report.py

@@ -17,7 +17,7 @@
 from polars.exceptions import ColumnNotFoundError
 
 from dve.pipeline.utils import SubmissionStatus
-from dve.reporting.constants import ErrorReportStatus
+from dve.reporting.constants import ErrorReportCategories, ErrorReportStatus
 
 
 @dataclass
@@ -92,18 +92,18 @@ def create_summary_sheet(
     def get_submission_status(self, aggregates: DataFrame) -> str:
         """Returns the status of the submission based on the error data"""
         if self.submission_status.processing_failed:
-            return "There was an issue processing the submission. Please contact support."
-        if self.submission_status.validation_failed:
-            return "File has been rejected"
+            return ErrorReportStatus.PROCESSING_FAILED
         if aggregates.is_empty():
-            return "File has been accepted, no issues to report"
+            return ErrorReportStatus.ACCEPTED
         failures = aggregates["Type"].unique()
-        if ErrorReportStatus.FILE_REJECTION.reporting_name in failures:
-            status = "File has been rejected"
-        elif ErrorReportStatus.WARNING.reporting_name in failures:
-            status = "File has been accepted, all records accepted with warnings"
+        if ErrorReportCategories.FILE_REJECTION.reporting_name in failures:
+            status = ErrorReportStatus.FILE_REJECTION
+        elif ErrorReportCategories.RECORD_REJECTION.reporting_name in failures:
+            status = ErrorReportStatus.RECORD_REJECTION
+        elif ErrorReportCategories.WARNING.reporting_name in failures:
+            status = ErrorReportStatus.ACCEPTED_WITH_WARNING
         else:
-            status = "File has been accepted, no issues to report"
+            status = ErrorReportStatus.ACCEPTED
         return status
 
     def _write_table(
@@ -153,6 +153,15 @@ def _add_submission_info(self, status: str, summary: Worksheet):
                 ),  # pylint: disable=C0301
             ]
         )
+        if status not in (
+            ErrorReportStatus.PROCESSING_FAILED,
+            ErrorReportStatus.FILE_REJECTION,
+        ):
+            summary.append([
+                "",
+                "Total Number of Records Rejected",
+                self.submission_status.number_of_records_rejected
+            ])
         summary.append(["", ""])
 
 

tests/test_pipeline/test_spark_pipeline.py

@@ -409,7 +409,7 @@ def test_error_report_where_report_is_expected(  # pylint: disable=redefined-out
     )
 
     submission_info, status, stats, report_uri = dve_pipeline.error_report(
-        sub_info, SubmissionStatus(True, 9)
+        sub_info, SubmissionStatus(True, 9, 2)
     )
 
     assert status.validation_failed
@@ -434,12 +434,13 @@ def test_error_report_where_report_is_expected(  # pylint: disable=redefined-out
         .rows()
     )
     assert report_records == [
-        ("Status", "File has been rejected"),
+        ("Status", "File has been accepted with record rejections"),
         ("Submission Id", submission_info.submission_id),
         ("Dataset Id", "planets"),
         ("File Name", "doesnotmatter"),
         ("File Extension", "json"),
         ("Total Number of Records Processed", "9"),
+        ("Total Number of Records Rejected", "2"),
         ("File Rejection", "0"),
         ("Record Rejection", "2"),
         ("Warning", "0"),

tests/test_reporting/test_excel_report.py

@@ -196,6 +196,3 @@ def test_sub_status_failed_processing():
         table_columns=["Planet", "Derived"],
     )
     assert summary_items.get_submission_status(pl.DataFrame()) == "There was an issue processing the submission. Please contact support."
-    summary_items.submission_status = SubmissionStatus(validation_failed=True)
-    assert summary_items.get_submission_status(pl.DataFrame()) == "File has been rejected"
-