From 6ecfc3d54b1d1f0fe82f5122cec0df690cf33df3 Mon Sep 17 00:00:00 2001 From: georgeRobertson <50412379+georgeRobertson@users.noreply.github.com> Date: Wed, 19 Aug 2026 18:19:20 +0100 Subject: [PATCH] fix: add user feedback message for unsupported file type --- src/dve/core_engine/models.py | 4 ++- src/dve/pipeline/pipeline.py | 6 ++-- src/dve/pipeline/utils.py | 35 ++++++++++++++++++--- tests/features/planets.feature | 4 ++- tests/test_pipeline/test_pipeline_utils.py | 36 ++++++++++++++++++++++ 5 files changed, 75 insertions(+), 10 deletions(-) create mode 100644 tests/test_pipeline/test_pipeline_utils.py diff --git a/src/dve/core_engine/models.py b/src/dve/core_engine/models.py index bba2986..49dba23 100644 --- a/src/dve/core_engine/models.py +++ b/src/dve/core_engine/models.py @@ -82,7 +82,9 @@ def _ensure_just_file_stem( @property def file_name_with_ext(self): """Return file name with extension.""" - return f"{self.file_name}.{self.file_extension}" + if self.file_extension: + return f"{self.file_name}.{self.file_extension}" + return self.file_name @classmethod def from_metadata_file(cls, submission_id: str, metadata_uri: Location): diff --git a/src/dve/pipeline/pipeline.py b/src/dve/pipeline/pipeline.py index a9be3ff..ca409dc 100644 --- a/src/dve/pipeline/pipeline.py +++ b/src/dve/pipeline/pipeline.py @@ -215,10 +215,10 @@ def write_file_to_parquet( for model_name, model in models.items(): self._logger.info(f"Transforming {model_name} to stringified parquet") - reader: BaseFileReader = load_reader( - dataset, model_name, ext, self.backend_reader_kwargs - ) try: + reader: BaseFileReader = load_reader( + dataset, model_name, ext, self.backend_reader_kwargs + ) if not entity_type: reader.write_parquet( reader.read_to_py_iterator( diff --git a/src/dve/pipeline/utils.py b/src/dve/pipeline/utils.py index e6122c2..be8ab66 100644 --- a/src/dve/pipeline/utils.py +++ b/src/dve/pipeline/utils.py @@ -11,9 +11,11 @@ import dve.core_engine.backends.implementations.duckdb # pylint: disable=unused-import import dve.core_engine.backends.implementations.spark # pylint: disable=unused-import import dve.parser.file_handling as fh +from dve.core_engine.backends.exceptions import MessageBearingError from dve.core_engine.backends.readers import _READER_REGISTRY from dve.core_engine.configuration.v1 import SchemaName, V1EngineConfig, _ModelConfig from dve.core_engine.loggers import get_logger +from dve.core_engine.message import FeedbackMessage from dve.core_engine.type_hints import URI, SubmissionResult from dve.metadata_parser.model_generator import JSONtoPyd @@ -52,11 +54,34 @@ def load_reader( backend_reader_kwargs: Optional[dict[str, Any]] = None, ): """Loads the readers for the diven feed, model name and file extension""" - reader_config = dataset[model_name].reader_config[f".{file_extension.lower()}"] - reader = _READER_REGISTRY[reader_config.reader]( - **reader_config.kwargs_, **backend_reader_kwargs if backend_reader_kwargs else {} - ) - return reader + try: + reader_config = dataset[model_name].reader_config[f".{file_extension.lower()}"] + reader = _READER_REGISTRY[reader_config.reader]( + **reader_config.kwargs_, **backend_reader_kwargs if backend_reader_kwargs else {} + ) + return reader + except KeyError as exc: + if file_extension: + err_msg = ( + f"The supplied file extension `{file_extension if file_extension else None}`" + +f" is not a supported file format for {model_name}." + ) + else: + err_msg = "No supplied file extension. Unable to parse file without a file extension." + + raise MessageBearingError( + "The file extension provided is not supported.", + messages=[ + FeedbackMessage( + entity=model_name, + record=None, + failure_type="submission", + error_location="Whole File", + error_code="InvalidFileExtension", + error_message=err_msg, + ) + ], + ) from exc def unpersist_all_rdds(spark: SparkSession): diff --git a/tests/features/planets.feature b/tests/features/planets.feature index b37b60b..0c4b21d 100644 --- a/tests/features/planets.feature +++ b/tests/features/planets.feature @@ -43,7 +43,9 @@ Feature: Pipeline tests using the planets dataset And I add initial audit entries for the submission Then the latest audit record for the submission is marked with processing status file_transformation When I run the file transformation phase - Then the latest audit record for the submission is marked with processing status failed + Then the latest audit record for the submission is marked with processing status error_report + When I run the error report phase + Then An error report is produced Scenario: Handle a file with duplicated extension provided (spark) Given I submit the planets file planets.csv.csv for processing diff --git a/tests/test_pipeline/test_pipeline_utils.py b/tests/test_pipeline/test_pipeline_utils.py new file mode 100644 index 0000000..fc28306 --- /dev/null +++ b/tests/test_pipeline/test_pipeline_utils.py @@ -0,0 +1,36 @@ +from dve.core_engine.backends.exceptions import MessageBearingError +from dve.core_engine.configuration.v1 import _ModelConfig, _ReaderConfig +from dve.pipeline.utils import load_reader + +import pytest + + +class TestLoadReader: + test_model_config = _ModelConfig( + fields={"test": "str"}, + reporting_fields=["test"], + key_field="test", + reader_config={ + ".csv": _ReaderConfig(reader="TestCsvReader"), + } + ) + + def test_invalid_load_reader_with_file_ext(self): + with pytest.raises(MessageBearingError) as exc_info: + load_reader( + {"test": self.test_model_config}, + "test_model", + "jpeg" + ) + + assert exc_info.value.messages[0].error_message == "The supplied file extension `jpeg` is not a supported file format for test_model." + + def test_invalid_load_reader_missing_file_ext(self): + with pytest.raises(MessageBearingError) as exc_info: + load_reader( + {"test": self.test_model_config}, + "test_model", + "" + ) + + assert exc_info.value.messages[0].error_message == "No supplied file extension. Unable to parse file without a file extension."