feat: some formatting. Tweaked how errors are handled within file transformation

Author: stevenhsd

src/dve/core_engine/exceptions.py

@@ -1,7 +1,6 @@
 """Exceptions emitted by the pipeline."""
 
 from collections.abc import Iterator
-from typing import Any
 
 from dve.core_engine.backends.implementations.spark.types import SparkEntities
 from dve.core_engine.message import FeedbackMessage
@@ -26,16 +25,17 @@ def __init__(
     def critical_messages(self) -> Iterator[FeedbackMessage]:
         """Critical messages which caused the processing error."""
         yield from filter(lambda message: message.is_critical, self.messages)
-    
+
     def to_feedback_message(self) -> FeedbackMessage:
+        "Convert to feedback message to write to json file"
         return FeedbackMessage(
             entity=None,
             record=None,
             failure_type="integrity",
             error_type="processing",
             error_location="Whole File",
-            error_message=self.error_message
-            )
+            error_message=self.error_message,
+        )
 
 
 class EntityTypeMismatch(TypeError):

src/dve/core_engine/message.py

@@ -445,6 +445,7 @@ def to_dict(
                 self.to_row(key_field, max_number_of_values, value_separator, record_converter),
             )
         )
+
     def __hash__(self):
         return hash(str(self))
 

src/dve/pipeline/duckdb_pipeline.py

@@ -40,7 +40,7 @@ def __init__(
             processed_files_path,
             submitted_files_path,
             reference_data_loader,
-            job_run_id
+            job_run_id,
         )
 
     # pylint: disable=arguments-differ
@@ -50,4 +50,3 @@ def write_file_to_parquet(  # type: ignore
         return super().write_file_to_parquet(
             submission_file_uri, submission_info, output, DuckDBPyRelation
         )
-    

src/dve/pipeline/foundry_ddb_pipeline.py

@@ -1,56 +1,57 @@
 """A duckdb pipeline for running on Foundry platform"""
-from typing import List, Optional, Tuple
-from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import duckdb_write_parquet
+
+from typing import Optional
+from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import duckdb_get_entity_count, duckdb_write_parquet
 from dve.core_engine.backends.utilities import dump_errors
 from dve.core_engine.models import SubmissionInfo
 from dve.core_engine.type_hints import URI, Failed
 from dve.pipeline.duckdb_pipeline import DDBDVEPipeline
 from dve.pipeline.utils import SubmissionStatus
 from dve.parser import file_handling as fh
 
-@duckdb_write_parquet
 class FoundryDDBPipeline(DDBDVEPipeline):
     """DuckDB pipeline for running on Foundry Platform"""
+
     def persist_audit_records(self, submission_info: SubmissionInfo) -> URI:
         """Write out key audit relations to parquet for persisting to datasets"""
         write_to = fh.joinuri(self.processed_files_path, submission_info.submission_id, "audit/")
         self.write_parquet(
             self._audit_tables._processing_status.get_relation(),
-            write_to + "processing_status.parquet")
+            write_to + "processing_status.parquet",
+        )
         self.write_parquet(
             self._audit_tables._submission_statistics.get_relation(),
-            write_to + "submission_statistics.parquet")
+            write_to + "submission_statistics.parquet",
+        )
         return write_to
-    
-    def file_transformation(self, submission_info: SubmissionInfo) -> SubmissionInfo | dict[str, str]:
+
+    def file_transformation(
+        self, submission_info: SubmissionInfo
+    ) -> SubmissionInfo | dict[str, str]:
         try:
             return super().file_transformation(submission_info)
-        except Exception as exc:
+        except Exception as exc: # pylint: disable=W0718
             self._logger.error(f"File transformation raised exception: {exc}")
             self._logger.exception(exc)
-            # TODO: write errors to file here (maybe processing errors - not to be seen by end user)
             return submission_info.dict()
-    
-    def apply_data_contract(self, submission_info: SubmissionInfo) -> Tuple[SubmissionInfo | bool]:
+
+    def apply_data_contract(self, submission_info: SubmissionInfo) -> tuple[SubmissionInfo | bool]:
         try:
             return super().apply_data_contract(submission_info)
-        except Exception as exc:
+        except Exception as exc: # pylint: disable=W0718
             self._logger.error(f"Apply data contract raised exception: {exc}")
             self._logger.exception(exc)
-            # TODO: write errors to file here (maybe processing errors - not to be seen by end user)
             return submission_info, True
-    
+
     def apply_business_rules(self, submission_info: SubmissionInfo, failed: Failed):
         try:
             return super().apply_business_rules(submission_info, failed)
-        except Exception as exc:
+        except Exception as exc: # pylint: disable=W0718
             self._logger.error(f"Apply business rules raised exception: {exc}")
             self._logger.exception(exc)
-            # TODO: write errors to file here (maybe processing errors - not to be seen by end user)
             return submission_info, SubmissionStatus(failed=True)
-            
-       
-    def run_pipeline(self, submission_info: SubmissionInfo) -> Tuple[Optional[URI], URI, URI]:
+
+    def run_pipeline(self, submission_info: SubmissionInfo) -> tuple[Optional[URI], URI, URI]:
         """Sequential single submission pipeline runner"""
         try:
             sub_id: str = submission_info.submission_id
@@ -61,23 +62,31 @@ def run_pipeline(self, submission_info: SubmissionInfo) -> Tuple[Optional[URI],
                 self._audit_tables.mark_data_contract(submission_ids=[sub_id])
                 sub_info, failed = self.apply_data_contract(submission_info=submission_info)
                 self._audit_tables.mark_business_rules(submissions=[(sub_id, failed)])
-                sub_info, sub_status = self.apply_business_rules(submission_info=submission_info, failed= failed)
+                sub_info, sub_status = self.apply_business_rules(
+                    submission_info=submission_info, failed=failed
+                )
             else:
-                sub_status = SubmissionStatus(failed=True)    
-            self._audit_tables.mark_error_report(submissions=[(sub_id, sub_status.submission_result)])
-            sub_info, sub_status, sub_stats, report_uri = self.error_report(submission_info=submission_info, status=sub_status)
+                sub_status = SubmissionStatus(failed=True)
+            self._audit_tables.mark_error_report(
+                submissions=[(sub_id, sub_status.submission_result)]
+            )
+            sub_info, sub_status, sub_stats, report_uri = self.error_report(
+                submission_info=submission_info, status=sub_status
+            )
             self._audit_tables.add_submission_statistics_records(sub_stats=[sub_stats])
-        except Exception as err:
-            self._logger.error(f"During processing of submission_id: {sub_id}, the following exception was raised: {err}")
+        except Exception as err: # pylint: disable=W0718
+            self._logger.error(
+                f"During processing of submission_id: {sub_id}, the following exception was raised: {err}"
+            )
             self._audit_tables.mark_failed(submissions=[sub_id])
         finally:
             audit_files_uri = self.persist_audit_records(submission_info=submission_info)
-            return  (
-                None if sub_status.failed else fh.joinuri(
-                    self.processed_files_path,
-                    sub_id,
-                    "business_rules"),
+            return (
+                (
+                    None
+                    if sub_status.failed
+                    else fh.joinuri(self.processed_files_path, sub_id, "business_rules")
+                ),
                 report_uri,
-                audit_files_uri
+                audit_files_uri,
             )
-            

src/dve/pipeline/pipeline.py

@@ -21,7 +21,11 @@
 from dve.core_engine.backends.base.core import EntityManager
 from dve.core_engine.backends.base.reference_data import BaseRefDataLoader
 from dve.core_engine.backends.base.rules import BaseStepImplementations
-from dve.core_engine.backends.exceptions import BackendError, MessageBearingError, ReaderLacksEntityTypeSupport
+from dve.core_engine.backends.exceptions import (
+    BackendError,
+    MessageBearingError,
+    ReaderLacksEntityTypeSupport,
+)
 from dve.core_engine.backends.readers import BaseFileReader
 from dve.core_engine.backends.types import EntityType
 from dve.core_engine.backends.utilities import dump_errors, stringify_model
@@ -52,7 +56,7 @@ def __init__(
         processed_files_path: Optional[URI],
         submitted_files_path: Optional[URI],
         reference_data_loader: Optional[type[BaseRefDataLoader]] = None,
-        job_run_id: Optional[int] = None
+        job_run_id: Optional[int] = None,
     ):
         self._submitted_files_path = submitted_files_path
         self._processed_files_path = processed_files_path
@@ -267,33 +271,41 @@ def file_transformation(
         if not self.processed_files_path:
             raise AttributeError("processed files path not provided")
 
+        errors: list[FeedbackMessage] = []
         submission_file_uri: URI = fh.joinuri(
             self.processed_files_path,
             submission_info.submission_id,
             submission_info.file_name_with_ext,
         )
         try:
-            errors = self.write_file_to_parquet(
+            errors.extend(self.write_file_to_parquet(
                 submission_file_uri, submission_info, self.processed_files_path
-            )
+            ))
+        
+        except MessageBearingError as exc:
+            self._logger.error(f"Unexpected file transformation error: {exc}")
+            self._logger.exception(exc)
+            errors.extend(exc.messages)
 
-        except Exception as exc:  # pylint: disable=broad-except
+        except BackendError as exc:  # pylint: disable=broad-except
             self._logger.error(f"Unexpected file transformation error: {exc}")
             self._logger.exception(exc)
-             # TODO: should this go to processing_errors.json?
-             # TODO: shouldn't be seen by user and don't need to maintain feedback message structure
-            errors = [CriticalProcessingError(entities=None,
-                                              error_message=repr(exc),
-                                              messages=[]).to_feedback_message()]
-        finally:
-            if errors:
-                dump_errors(
-                    fh.joinuri(self.processed_files_path, submission_info.submission_id),
-                    "file_transformation",
-                    errors,
-                )
-                return submission_info.dict()
-            return submission_info
+            errors.extend([
+                CriticalProcessingError(
+                    entities=None,
+                    error_message=repr(exc),
+                    messages=[],
+                ).to_feedback_message()
+            ])
+
+        if errors:
+            dump_errors(
+                fh.joinuri(self.processed_files_path, submission_info.submission_id),
+                "file_transformation",
+                errors,
+            )
+            return submission_info.dict()
+        return submission_info
 
     def file_transformation_step(
         self, pool: Executor, submissions_to_process: list[SubmissionInfo]
@@ -326,7 +338,7 @@ def file_transformation_step(
             except Exception as exc:  # pylint: disable=W0703
                 self._logger.error(f"File transformation raised exception: {exc}")
                 self._logger.exception(exc)
-                # TODO: write errors to file here (maybe processing errors - not to be seen by end user) 
+                # TODO: write errors to file here (maybe processing errors - not to be seen by end user)
                 failed_processing.append(sub_info)
                 continue
 

src/dve/pipeline/spark_pipeline.py

@@ -42,7 +42,7 @@ def __init__(
             processed_files_path,
             submitted_files_path,
             reference_data_loader,
-            job_run_id
+            job_run_id,
         )
 
     # pylint: disable=arguments-differ