From f0ddad3e9a3348e6af449118cb1879c39331cd25 Mon Sep 17 00:00:00 2001 From: "vaibhav.tiwari" <249828730+vaibhavtiwari16@users.noreply.github.com> Date: Tue, 13 Oct 2026 12:14:38 +0100 Subject: [PATCH] Add error_report config to toggle record rejection reporting --- .../core_engine/configuration/v1/__init__.py | 8 +++++ src/dve/pipeline/pipeline.py | 29 +++++++++++++++---- tests/features/animals.feature | 4 +-- tests/testdata/animals/animals.dischema.json | 3 ++ 4 files changed, 36 insertions(+), 8 deletions(-) diff --git a/src/dve/core_engine/configuration/v1/__init__.py b/src/dve/core_engine/configuration/v1/__init__.py index 959596fc..fc0bfbbd 100644 --- a/src/dve/core_engine/configuration/v1/__init__.py +++ b/src/dve/core_engine/configuration/v1/__init__.py @@ -166,6 +166,12 @@ 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.""" @@ -173,6 +179,8 @@ class V1EngineConfig(BaseEngineConfig): """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 ) diff --git a/src/dve/pipeline/pipeline.py b/src/dve/pipeline/pipeline.py index a9be3ff9..bbfe6ae8 100644 --- a/src/dve/pipeline/pipeline.py +++ b/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 @@ -34,6 +33,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.exceptions import CriticalProcessingError from dve.core_engine.loggers import get_logger from dve.core_engine.message import FeedbackMessage @@ -745,8 +745,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") @@ -794,12 +793,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] ]: @@ -821,7 +827,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 @@ -852,7 +864,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( diff --git a/tests/features/animals.feature b/tests/features/animals.feature index d68ddbf1..1a2efc17 100644 --- a/tests/features/animals.feature +++ b/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 | diff --git a/tests/testdata/animals/animals.dischema.json b/tests/testdata/animals/animals.dischema.json index 0e1eda14..7180d093 100644 --- a/tests/testdata/animals/animals.dischema.json +++ b/tests/testdata/animals/animals.dischema.json @@ -50,5 +50,8 @@ "is_informational": true } ] + }, + "error_report": { + "record_rejection_reporting": false } } \ No newline at end of file