refactor: small changes to foundry pipeline and duckdb csv to fix header check

Author: stevenhsd

src/dve/core_engine/backends/implementations/duckdb/readers/csv.py

@@ -20,7 +20,7 @@
 from dve.core_engine.backends.utilities import get_polars_type_from_annotation
 from dve.core_engine.message import FeedbackMessage
 from dve.core_engine.type_hints import URI, EntityName
-from dve.parser.file_handling import get_content_length, open_stream
+from dve.parser.file_handling import get_content_length
 
 
 @duckdb_write_parquet
@@ -38,7 +38,7 @@ def __init__(
         connection: Optional[DuckDBPyConnection] = None,
         field_check: bool = False,
         field_check_error_code: Optional[str] = "ExpectedVsActualFieldMismatch",
-        field_check_error_message: Optional[str] = "The submitted header does not match what is expected",
+        field_check_error_message: Optional[str] = "The submitted header is missing fields",
         **_,
     ):
         self.header = header
@@ -67,11 +67,11 @@ def perform_field_check(
                 messages=[
                     FeedbackMessage(
                         entity=entity_name,
+                        record=None,
                         failure_type="submission",
                         error_location="Whole File",
                         error_code=self.field_check_error_code,
-                        error_message=self.field_check_error_message,
-                        value=f"Missing fields: {missing}",
+                        error_message=f"{self.field_check_error_message} - missing fields: {missing}",
                     )
                 ],
             )

src/dve/pipeline/foundry_ddb_pipeline.py

@@ -109,6 +109,7 @@ def error_report(
             self._logger.exception(exc)
             sub_stats = None
             report_uri = None
+            submission_status.processing_failed = True
             dump_processing_errors(
                 fh.joinuri(self.processed_files_path, submission_info.submission_id),
                 "error_report",
@@ -148,7 +149,8 @@ def run_pipeline(
                 sub_info, sub_status, sub_stats, report_uri = self.error_report(
                     submission_info=submission_info, submission_status=sub_status
                 )
-                self._audit_tables.add_submission_statistics_records(sub_stats=[sub_stats])
+                if sub_stats:
+                    self._audit_tables.add_submission_statistics_records(sub_stats=[sub_stats])
         except Exception as err:  # pylint: disable=W0718
             self._logger.error(
                 f"During processing of submission_id: {sub_id}, this exception was raised: {err}"