diff --git a/src/dve/core_engine/configuration/v1/__init__.py b/src/dve/core_engine/configuration/v1/__init__.py index 634cc112..4f3b4115 100644 --- a/src/dve/core_engine/configuration/v1/__init__.py +++ b/src/dve/core_engine/configuration/v1/__init__.py @@ -248,6 +248,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.""" @@ -255,6 +261,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 2cf00553..35225c1a 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 @@ -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( 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