feat: add error_report config to toggle record rejection reporting (#185)

Author: georgeRobertson

src/dve/core_engine/configuration/v1/__init__.py

@@ -248,13 +248,21 @@ class V1TransformationConfig(BaseModel):
     """Complex rules."""
 
 
+class V1ErrorReportConfig(BaseModel):
+    """Configuration for the error report component of the dataset."""
+
+    record_rejection_reporting: bool = True
+
+
 class V1EngineConfig(BaseEngineConfig):
     """A loader for the first version of the configuration language."""
 
     contract: V1DataContractConfig
     """The data contract configuration."""
     transformations: V1TransformationConfig = Field(default_factory=V1TransformationConfig)
     """The transformation/rules configuration."""
+    error_report: V1ErrorReportConfig = Field(default_factory=V1ErrorReportConfig)
+    """The error report configuration."""
     _rule_store_rules: dict[RuleName, BusinessComponentSpecConfigUnion] = PrivateAttr(
         default_factory=dict
     )

src/dve/pipeline/pipeline.py

@@ -7,7 +7,6 @@
 from collections import defaultdict
 from collections.abc import Generator, Iterable, Iterator
 from concurrent.futures import Executor, Future, ThreadPoolExecutor
-from functools import lru_cache
 from itertools import starmap
 from threading import Lock
 from typing import Any, Optional, Union
@@ -35,6 +34,7 @@
 from dve.core_engine.backends.readers.utilities import get_all_model_fields
 from dve.core_engine.backends.types import EntityType
 from dve.core_engine.backends.utilities import stringify_model
+from dve.core_engine.configuration.v1 import V1ErrorReportConfig
 from dve.core_engine.configuration.v1.hierarchy import EntityHierarchy
 from dve.core_engine.exceptions import CriticalProcessingError
 from dve.core_engine.loggers import get_logger
@@ -886,8 +886,7 @@ def _publish_error_aggregates(self, submission_id: str, aggregates_df: pl.DataFr
         aggregates_df.write_parquet(output_uri)
         return output_uri
 
-    @lru_cache()  # noqa: B019
-    def _get_error_dataframes(self, submission_id: str):
+    def _get_error_dataframes(self, submission_id: str, include_record_rejections: bool = True):
         if not self.processed_files_path:
             raise AttributeError("processed files path not provided")
 
@@ -935,12 +934,19 @@ def _get_error_dataframes(self, submission_id: str):
                 errors_dfs.append(df)
 
         errors_df = pl.concat(errors_dfs, how="align")  # type: ignore
+        if not include_record_rejections:
+            errors_df = errors_df.filter(
+                pl.col("Type") != ErrorReportCategories.RECORD_REJECTION.reporting_name
+            )
         aggregates = calculate_aggregates(errors_df)
 
         return errors_df, aggregates
 
     def error_report(
-        self, submission_info: SubmissionInfo, submission_status: Optional[SubmissionStatus] = None
+        self,
+        submission_info: SubmissionInfo,
+        submission_status: Optional[SubmissionStatus] = None,
+        error_report_config: Optional[V1ErrorReportConfig] = None,
     ) -> tuple[
         SubmissionInfo, SubmissionStatus, Optional[SubmissionStatisticsRecord], Optional[URI]
     ]:
@@ -962,7 +968,13 @@ def error_report(
             raise AttributeError("processed files path not provided")
 
         self._logger.info("Reading error dataframes")
-        errors_df, aggregates = self._get_error_dataframes(submission_info.submission_id)
+        if error_report_config is None:
+            _, config, _ = load_config(submission_info.dataset_id, self.rules_path)
+            error_report_config = config.error_report
+        errors_df, aggregates = self._get_error_dataframes(
+            submission_info.submission_id,
+            error_report_config.record_rejection_reporting,
+        )
 
         if not submission_status.number_of_records:
             sub_stats = None
@@ -993,7 +1005,12 @@ def error_report(
         summary_items = er.SummaryItems(
             submission_status=submission_status,
             summary_dict=summary_dict,
-            row_headings=[e.reporting_name for e in ErrorReportCategories],
+            row_headings=[
+                e.reporting_name
+                for e in ErrorReportCategories
+                if error_report_config.record_rejection_reporting
+                or e != ErrorReportCategories.RECORD_REJECTION
+            ],
         )
 
         workbook = er.ExcelFormat(

tests/features/animals.feature

@@ -24,7 +24,7 @@ Feature: Pipeline tests using the animal dataset
         And The statistics entry for the submission shows the following information
             | parameter                | value |
             | record_count             | 5     |
-            | number_record_rejections | 2     |
+            | number_record_rejections | 0     |
             | number_warnings          | 0     |
 
     Scenario: Validate XML data with a mixture of error types in (duckdb)
@@ -55,5 +55,5 @@ Feature: Pipeline tests using the animal dataset
             | parameter                    | value |
             | record_count                 | 7     |
             | number_submission_rejections | 1     |
-            | number_record_rejections     | 2     |
+            | number_record_rejections     | 0     |
             | number_warnings              | 1     |

tests/testdata/animals/animals.dischema.json

@@ -50,5 +50,8 @@
                 "is_informational": true
             }
         ]
+    },
+    "error_report": {
+        "record_rejection_reporting": false
     }
 }