77from collections import defaultdict
88from collections .abc import Generator , Iterable , Iterator
99from concurrent .futures import Executor , Future , ThreadPoolExecutor
10- from functools import lru_cache
1110from itertools import starmap
1211from threading import Lock
1312from typing import Any , Optional , Union
3433from dve .core_engine .backends .readers .utilities import get_all_model_fields
3534from dve .core_engine .backends .types import EntityType
3635from dve .core_engine .backends .utilities import stringify_model
36+ from dve .core_engine .configuration .v1 import V1ErrorReportConfig
3737from dve .core_engine .exceptions import CriticalProcessingError
3838from dve .core_engine .loggers import get_logger
3939from dve .core_engine .message import FeedbackMessage
@@ -745,8 +745,7 @@ def _publish_error_aggregates(self, submission_id: str, aggregates_df: pl.DataFr
745745 aggregates_df .write_parquet (output_uri )
746746 return output_uri
747747
748- @lru_cache () # noqa: B019
749- def _get_error_dataframes (self , submission_id : str ):
748+ def _get_error_dataframes (self , submission_id : str , include_record_rejections : bool = True ):
750749 if not self .processed_files_path :
751750 raise AttributeError ("processed files path not provided" )
752751
@@ -794,12 +793,19 @@ def _get_error_dataframes(self, submission_id: str):
794793 errors_dfs .append (df )
795794
796795 errors_df = pl .concat (errors_dfs , how = "align" ) # type: ignore
796+ if not include_record_rejections :
797+ errors_df = errors_df .filter (
798+ pl .col ("Type" ) != ErrorReportCategories .RECORD_REJECTION .reporting_name
799+ )
797800 aggregates = calculate_aggregates (errors_df )
798801
799802 return errors_df , aggregates
800803
801804 def error_report (
802- self , submission_info : SubmissionInfo , submission_status : Optional [SubmissionStatus ] = None
805+ self ,
806+ submission_info : SubmissionInfo ,
807+ submission_status : Optional [SubmissionStatus ] = None ,
808+ error_report_config : Optional [V1ErrorReportConfig ] = None ,
803809 ) -> tuple [
804810 SubmissionInfo , SubmissionStatus , Optional [SubmissionStatisticsRecord ], Optional [URI ]
805811 ]:
@@ -821,7 +827,13 @@ def error_report(
821827 raise AttributeError ("processed files path not provided" )
822828
823829 self ._logger .info ("Reading error dataframes" )
824- errors_df , aggregates = self ._get_error_dataframes (submission_info .submission_id )
830+ if error_report_config is None :
831+ _ , config , _ = load_config (submission_info .dataset_id , self .rules_path )
832+ error_report_config = config .error_report
833+ errors_df , aggregates = self ._get_error_dataframes (
834+ submission_info .submission_id ,
835+ error_report_config .record_rejection_reporting ,
836+ )
825837
826838 if not submission_status .number_of_records :
827839 sub_stats = None
@@ -852,7 +864,12 @@ def error_report(
852864 summary_items = er .SummaryItems (
853865 submission_status = submission_status ,
854866 summary_dict = summary_dict ,
855- row_headings = [e .reporting_name for e in ErrorReportCategories ],
867+ row_headings = [
868+ e .reporting_name
869+ for e in ErrorReportCategories
870+ if error_report_config .record_rejection_reporting
871+ or e != ErrorReportCategories .RECORD_REJECTION
872+ ],
856873 )
857874
858875 workbook = er .ExcelFormat (
0 commit comments