From 2b0d43fe613b7dd24dc2b9e8c1f72297e5bfe23c Mon Sep 17 00:00:00 2001 From: stevenhsd <56357022+stevenhsd@users.noreply.github.com> Date: Wed, 4 Nov 2026 16:20:05 +0000 Subject: [PATCH 1/2] fix: ensure reader errors are handled correctly in file transformation stage --- src/dve/core_engine/backends/exceptions.py | 48 +++++++++++++++---- .../implementations/duckdb/readers/csv.py | 4 +- .../implementations/spark/readers/csv.py | 2 +- .../implementations/spark/readers/json.py | 2 +- .../implementations/spark/readers/xml.py | 2 +- src/dve/core_engine/backends/readers/csv.py | 4 +- src/dve/core_engine/backends/readers/xml.py | 2 +- src/dve/core_engine/exceptions.py | 2 +- 8 files changed, 49 insertions(+), 17 deletions(-) diff --git a/src/dve/core_engine/backends/exceptions.py b/src/dve/core_engine/backends/exceptions.py index 99808c8a..552fa039 100644 --- a/src/dve/core_engine/backends/exceptions.py +++ b/src/dve/core_engine/backends/exceptions.py @@ -233,32 +233,55 @@ def get_message_preamble(self) -> EntityName: 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" - - -class MissingHeaderError(ReaderErrorMixin, ValueError): + + 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="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 ( @@ -267,7 +290,7 @@ def get_message_preamble(self) -> str: ) -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 +313,16 @@ def __init__( 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" diff --git a/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py b/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py index 488f8c85..4f74ad6f 100644 --- a/src/dve/core_engine/backends/implementations/duckdb/readers/csv.py +++ b/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) diff --git a/src/dve/core_engine/backends/implementations/spark/readers/csv.py b/src/dve/core_engine/backends/implementations/spark/readers/csv.py index 6b79bb35..73e1b1b1 100644 --- a/src/dve/core_engine/backends/implementations/spark/readers/csv.py +++ b/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) diff --git a/src/dve/core_engine/backends/implementations/spark/readers/json.py b/src/dve/core_engine/backends/implementations/spark/readers/json.py index 6404231c..e8fe90f2 100644 --- a/src/dve/core_engine/backends/implementations/spark/readers/json.py +++ b/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 = { diff --git a/src/dve/core_engine/backends/implementations/spark/readers/xml.py b/src/dve/core_engine/backends/implementations/spark/readers/xml.py index 769c480e..db1fa3f6 100644 --- a/src/dve/core_engine/backends/implementations/spark/readers/xml.py +++ b/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) diff --git a/src/dve/core_engine/backends/readers/csv.py b/src/dve/core_engine/backends/readers/csv.py index 619a3d19..be6b42db 100644 --- a/src/dve/core_engine/backends/readers/csv.py +++ b/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) diff --git a/src/dve/core_engine/backends/readers/xml.py b/src/dve/core_engine/backends/readers/xml.py index 05167e18..4e31de42 100644 --- a/src/dve/core_engine/backends/readers/xml.py +++ b/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): diff --git a/src/dve/core_engine/exceptions.py b/src/dve/core_engine/exceptions.py index 4877baf1..6a5adcc2 100644 --- a/src/dve/core_engine/exceptions.py +++ b/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 From 63a8d514f884803dffbcac9dd8260cd969771f23 Mon Sep 17 00:00:00 2001 From: stevenhsd <56357022+stevenhsd@users.noreply.github.com> Date: Wed, 4 Nov 2026 16:30:41 +0000 Subject: [PATCH 2/2] style: resolve linting issues --- src/dve/core_engine/backends/exceptions.py | 70 +++++++++++++--------- 1 file changed, 41 insertions(+), 29 deletions(-) diff --git a/src/dve/core_engine/backends/exceptions.py b/src/dve/core_engine/backends/exceptions.py index 552fa039..d873374e 100644 --- a/src/dve/core_engine/backends/exceptions.py +++ b/src/dve/core_engine/backends/exceptions.py @@ -238,15 +238,19 @@ class EmptyFileError(ReaderErrorMixin, CriticalMessageBearingError): def get_message_preamble(self) -> str: return "File location provided is empty" - + 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="EMPTYFILE", - error_message=error_message or self.get_message_preamble())) + + 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): @@ -254,14 +258,18 @@ class MissingHeaderError(ReaderErrorMixin, CriticalMessageBearingError): 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())) + 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, CriticalMessageBearingError): @@ -275,13 +283,17 @@ def __init__(self, *args: object, n_expected_fields: int, n_actual_fields: int) """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())) + + 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 ( @@ -316,13 +328,13 @@ def __init__( 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() - ) - ) + 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"