fix: ensure reader errors are handled correctly in file transformatio…

src/dve/core_engine/backends/exceptions.py

@@ -233,41 +233,76 @@
         return f"Refdata loader does not support reading refdata from {self.file_extension} files"
 
 
-class EmptyFileError(ReaderErrorMixin, ValueError):
+class EmptyFileError(ReaderErrorMixin, CriticalMessageBearingError):
     """The read file was empty."""
 
     def get_message_preamble(self) -> str:
         return "File location provided is empty"
 
+    def __init__(self, *args: object, error_message: Optional[str] = None):
 
-class MissingHeaderError(ReaderErrorMixin, ValueError):
+        super().__init__(
+            *args,
+            message=FeedbackMessage(
+                entity="Whole File",
+                record=None,
+                failure_type="submission",
+                error_code="EMPTYFILE",
+                error_message=error_message or self.get_message_preamble(),
+            ),
+        )
+
+
+class MissingHeaderError(ReaderErrorMixin, CriticalMessageBearingError):
     """A header is expected, but was not parsed."""
 
     def get_message_preamble(self) -> str:
         return "No header was provided"
 
+    def __init__(self, *args: object, error_message: Optional[str] = None):
+        super().__init__(
+            *args,
+            message=FeedbackMessage(
+                entity="Whole File",
+                record=None,
+                failure_type="submission",
+                error_code="MISSINGHEADER",
+                error_message=error_message or self.get_message_preamble(),
+            ),
+        )
+
 
-class FieldCountMismatch(ReaderErrorMixin, ValueError):
+class FieldCountMismatch(ReaderErrorMixin, CriticalMessageBearingError):
     """An error raised when the number of expected fields does not match
     the number of parsed fields.
 
     """
 
     def __init__(self, *args: object, n_expected_fields: int, n_actual_fields: int) -> None:
-        super().__init__(*args)
         self.n_expected_fields = n_expected_fields
         """The number of fields that should have been in the file."""
         self.n_actual_fields = n_actual_fields
         """The number of fields that were actually in the file."""
 
+        super().__init__(
+            *args,
+            message=FeedbackMessage(
+                entity="Whole File",
+                record=None,
+                failure_type="submission",
+                error_code="FIELDCOUNTMISMATCH",
+                error_message=self.get_message_preamble(),
+            ),
+        )
+
     def get_message_preamble(self) -> str:
         return (
             f"Number of fields expected ({self.n_expected_fields}) does not "
             + f"match the number in the provided data ({self.n_actual_fields})"
         )
 
 
-class SchemaMismatch(ReaderErrorMixin, ValueError):
+class SchemaMismatch(ReaderErrorMixin, CriticalMessageBearingError):
     """An error raised when a file does not match a given schema (e.g. fields
     are missing).
 
@@ -290,7 +325,16 @@
         tuple of actual and expected field type, both as strings).
 
         """
-        super().__init__(*args)
+        super().__init__(
+            *args,
+            message=FeedbackMessage(
+                entity="Whole File",
+                record=None,
+                failure_type="submission",
+                error_code="SCHEMAMISMATCH",
+                error_message=self.get_message_preamble() + self.get_message_epilogue(),
+            ),
+        )
 
     def get_message_preamble(self) -> str:
         return "Schema mismatch"

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

@@ -99,7 +99,7 @@ def read_to_relation(  # pylint: disable=unused-argument
     ) -> DuckDBPyRelation:
         """Returns a relation object from the source csv"""
         if get_content_length(resource) == 0:
-            raise EmptyFileError(f"File at {resource} is empty.")
+            raise EmptyFileError(error_message=f"File at {resource} is empty.")
 
         if self.field_check:
             self.perform_field_check(resource, entity_name, schema, all_model_fields)
@@ -155,7 +155,7 @@ def read_to_relation(  # pylint: disable=unused-argument
     ) -> DuckDBPyRelation:
         """Returns a relation object from the source csv"""
         if get_content_length(resource) == 0:
-            raise EmptyFileError(f"File at {resource} is empty.")
+            raise EmptyFileError(error_message=f"File at {resource} is empty.")
 
         if self.field_check:
             self.perform_field_check(resource, entity_name, schema, all_model_fields)

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

@@ -80,7 +80,7 @@ def read_to_dataframe(
     ) -> DataFrame:
         """Read a CSV file directly to a Spark DataFrame."""
         if get_content_length(resource) == 0:
-            raise EmptyFileError(f"File at {resource} is empty.")
+            raise EmptyFileError(error_message=f"File at {resource} is empty.")
 
         if self.field_check:
             self.perform_field_check(resource, entity_name, schema, all_model_fields)

src/dve/core_engine/backends/implementations/spark/readers/json.py

@@ -60,7 +60,7 @@ def read_to_dataframe(
     ) -> DataFrame:
         """Read a JSON file directly to a Spark DataFrame."""
         if get_content_length(resource) == 0:
-            raise EmptyFileError(f"File at {resource} is empty.")
+            raise EmptyFileError(error_message=f"File at {resource} is empty.")
 
         spark_schema: StructType = get_type_from_annotation(schema)
         kwargs = {

src/dve/core_engine/backends/implementations/spark/readers/xml.py

@@ -130,7 +130,7 @@ def read_to_dataframe(
 
         """
         if get_content_length(resource) == 0:
-            raise EmptyFileError(f"File at {resource} is empty.")
+            raise EmptyFileError(error_message=f"File at {resource} is empty.")
 
         if self.xsd_location:
             msg = self._run_xmllint(file_uri=resource)

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

@@ -130,7 +130,7 @@ def _parse_field_names(self, stream: IO[str]) -> list[str]:
         # can lead to null headers.
         parsed_field_names = reader.fieldnames
         if parsed_field_names is None:  # pragma: no cover
-            raise MissingHeaderError("Header could not be parsed from CSV file")
+            raise MissingHeaderError(error_message="Header could not be parsed from CSV file")
 
         if not all(
             map(lambda string: isinstance(string, str), parsed_field_names)
@@ -231,7 +231,7 @@ def read_to_py_iterator(
     ) -> Iterator[dict[str, Any]]:
         """Reads the data to an iterator of dictionaries"""
         if get_content_length(resource) == 0:
-            raise EmptyFileError(f"File at {resource!r} is empty")
+            raise EmptyFileError(error_message=f"File at {resource!r} is empty")
 
         if self.field_check:
             self.perform_field_check(resource, entity_name, schema, all_model_fields)

src/dve/core_engine/backends/readers/xml.py

@@ -319,7 +319,7 @@ def read_to_py_iterator(
 
         """
         if get_content_length(resource) == 0:
-            raise EmptyFileError(f"File at {resource!r} is empty")
+            raise EmptyFileError(error_message=f"File at {resource!r} is empty")
 
         with open_stream(resource, "rb") as stream:
             for idx, record in enumerate(self._parse_xml(stream, schema), start=1):

src/dve/core_engine/exceptions.py

@@ -13,7 +13,7 @@ def __init__(
         *args: object,
         messages: Optional[list[str]] = None,
     ) -> None:
-        super().__init__(error_message, *args)
+        super().__init__(*args)
         self.error_message = error_message
         """The error message explaining the critical processing error."""
         self.messages = messages