style: Additional logging around file transformation, business rules and error reports

Author: stevenhsd

src/dve/core_engine/backends/base/rules.py

@@ -360,13 +360,15 @@ def apply_sync_filters(
 
         messages: Messages = []
         for entity_name, filter_rules in filters_by_entity.items():
+            self.logger.info(f"Applying filters to {entity_name}")
             entity = entities[entity_name]
 
             filter_column_names: list[str] = []
             unmodified_entities = {entity_name: entity}
             modified_entities = {entity_name: entity}
 
             for rule in filter_rules:
+                self.logger.info(f"Applying filter {rule.reporting.code}")
                 if rule.reporting.emit == "record_failure":
                     column_name = f"filter_{uuid4().hex}"
                     filter_column_names.append(column_name)
@@ -412,6 +414,7 @@ def apply_sync_filters(
                         return messages, False
 
             if filter_column_names:
+                self.logger.info(f"Filtering records where validation is record level")
                 success_condition = " AND ".join(
                     [f"({c_name} IS NOT NULL AND {c_name})" for c_name in filter_column_names]
                 )
@@ -456,6 +459,7 @@ def apply_rules(self, entities: Entities, rule_metadata: RuleMetadata) -> Messag
         altering the entities in-place.
 
         """
+        self.logger.info("Applying business rules")
         rules_and_locals: Iterable[tuple[Rule, TemplateVariables]]
         if rule_metadata.templating_strategy == "upfront":
             rules_and_locals = []
@@ -472,6 +476,8 @@ def apply_rules(self, entities: Entities, rule_metadata: RuleMetadata) -> Messag
             rules_and_locals = rule_metadata
 
         messages: Messages = []
+        
+        self.logger.info("Applying pre-sync steps")
         for rule, local_variables in rules_and_locals:
             for step in rule.pre_sync_steps:
                 if rule_metadata.templating_strategy == "runtime":
@@ -498,6 +504,8 @@ def apply_rules(self, entities: Entities, rule_metadata: RuleMetadata) -> Messag
         if not success:
             return messages
 
+        self.logger.info("Applying post-sync steps")
+        
         for rule, local_variables in rules_and_locals:
             for step in rule.post_sync_steps:
                 if rule_metadata.templating_strategy == "runtime":

src/dve/pipeline/pipeline.py

@@ -190,6 +190,7 @@ def write_file_to_parquet(
         errors = []
 
         for model_name, model in models.items():
+            self._logger.info(f"Transforming {model_name} to stringly typed parquet")
             reader: BaseFileReader = load_reader(dataset, model_name, ext)
             try:
                 if not entity_type:
@@ -747,7 +748,7 @@ def error_report(
         SubmissionInfo, SubmissionStatus, Optional[SubmissionStatisticsRecord], Optional[URI]
     ]:
         """Creates the error reports given a submission info and submission status"""
-
+        self._logger.info("Generating error report")
         if not submission_status:
             submission_status = self.get_submission_status(
                 "error_report", submission_info.submission_id
@@ -756,6 +757,7 @@ def error_report(
         if not self.processed_files_path:
             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 not submission_status.number_of_records:
@@ -794,9 +796,11 @@ def error_report(
             "error_reports",
             f"{submission_info.file_name}_{submission_info.file_extension.strip('.')}.xlsx",
         )
+        self._logger.info("Writing error report")
         with fh.open_stream(report_uri, "wb") as stream:
             stream.write(er.ExcelFormat.convert_to_bytes(workbook))
 
+        self._logger.info("Publishing error aggregates")
         self._publish_error_aggregates(submission_info.submission_id, aggregates)
 
         return submission_info, submission_status, sub_stats, report_uri