refactor: merge in main and resolve conflicts and linting issues

Author: stevenhsd

CHANGELOG.md

@@ -1,3 +1,18 @@
+## v0.5.2 (2026-02-02)
+
+### Refactor
+
+- allow passing of custom loggers into pipeline objects
+- ensure traceback in broad exceptions
+- improve the logging around dve processing errors and align reporting to module name rather than legacy name
+- add sense check for text based file (#32)
+
+## v0.5.1 (2026-01-28)
+
+### Fix
+
+- deal with pathing assumption that file had been moved to processed_file_path during file transformation
+
 ## v0.5.0 (2026-01-16)
 
 ### Feat

Makefile

@@ -17,7 +17,7 @@ behave:
 	${activate} behave
 
 pytest:
-	${activate} pytest tests/
+	${activate} pytest -c pytest-dev.ini
 
 all-tests: pytest behave
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "nhs_dve"
-version = "0.5.0"
+version = "0.5.2"
 description = "`nhs data validation engine` is a framework used to validate data"
 authors = ["NHS England <england.contactus@nhs.net>"]
 readme = "README.md"

pytest-dev.ini

@@ -0,0 +1,3 @@
+[pytest]
+log_cli = true
+log_cli_level = INFO

src/dve/common/error_utils.py

@@ -66,8 +66,8 @@ def dump_feedback_errors(
 
 
 def dump_processing_errors(
-    working_folder: URI, step_name: DVEStage, errors: Union[list[CriticalProcessingError], Messages]
-) -> URI:
+    working_folder: URI, step_name: str, errors: list[CriticalProcessingError]
+):
     """Write out critical processing errors"""
     if not working_folder:
         raise AttributeError("processed files path not passed")
@@ -76,27 +76,17 @@ def dump_processing_errors(
     if not errors:
         raise AttributeError("errors list not passed")
 
-    error_file: URI = get_processing_errors_uri(working_folder)
+    error_file: URI = fh.joinuri(working_folder, "processing_errors", "processing_errors.json")
     processed = []
 
     for error in errors:
-        if isinstance(error, CriticalProcessingError):
-            if msgs := error.messages:
-                for msg in msgs:
-                    processed.append(
-                        {
-                            "step_name": step_name,
-                            "error_location": msg.error_location,
-                            "error_level": msg.error_type,
-                            "error_message": msg.error_message,
-                        }
-                    )
         processed.append(
             {
                 "step_name": step_name,
                 "error_location": "processing",
                 "error_level": "integrity",
                 "error_message": error.error_message,
+                "error_traceback": error.messages,
             }
         )
 

src/dve/core_engine/backends/base/contract.py

@@ -21,6 +21,7 @@
 from dve.core_engine.backends.readers import get_reader
 from dve.core_engine.backends.types import Entities, EntityType, StageSuccessful
 from dve.core_engine.backends.utilities import dedup_messages, stringify_model
+from dve.core_engine.exceptions import CriticalProcessingError
 from dve.core_engine.loggers import get_logger
 from dve.core_engine.message import FeedbackMessage
 from dve.core_engine.type_hints import (
@@ -395,7 +396,16 @@ def apply(
         processing_errors_uri = get_processing_errors_uri(working_dir)
         entities, messages, successful = self.read_raw_entities(entity_locations, contract_metadata)
         if not successful:
-            dump_processing_errors(working_dir, "data_contract", messages)
+            dump_processing_errors(
+                working_dir,
+                "data_contract",
+                [
+                    CriticalProcessingError(
+                        "Issue occurred while reading raw entities",
+                        [msg.error_message for msg in messages],
+                    )
+                ],
+            )
             return {}, feedback_errors_uri, successful, processing_errors_uri
 
         try:
@@ -409,7 +419,16 @@ def apply(
                 "data contract",
                 self.logger,
             )
-            dump_processing_errors(working_dir, "data_contract", new_messages)
+            dump_processing_errors(
+                working_dir,
+                "data_contract",
+                [
+                    CriticalProcessingError(
+                        "Issue occurred while applying data_contract",
+                        [msg.error_message for msg in new_messages],
+                    )
+                ],
+            )
 
         if contract_metadata.cache_originals:
             for entity_name in list(entities):

src/dve/core_engine/backends/base/reader.py

@@ -8,9 +8,11 @@
 from pydantic import BaseModel
 from typing_extensions import Protocol
 
-from dve.core_engine.backends.exceptions import ReaderLacksEntityTypeSupport
+from dve.core_engine.backends.exceptions import MessageBearingError, ReaderLacksEntityTypeSupport
 from dve.core_engine.backends.types import EntityName, EntityType
+from dve.core_engine.message import FeedbackMessage
 from dve.core_engine.type_hints import URI, ArbitraryFunction, WrapDecorator
+from dve.parser.file_handling.service import open_stream
 
 T = TypeVar("T")
 ET_co = TypeVar("ET_co", covariant=True)
@@ -116,6 +118,8 @@ def read_to_entity_type(
         if entity_name == Iterator[dict[str, Any]]:
             return self.read_to_py_iterator(resource, entity_name, schema)  # type: ignore
 
+        self.raise_if_not_sensible_file(resource, entity_name)
+
         try:
             reader_func = self.__read_methods__[entity_type]
         except KeyError as err:
@@ -137,3 +141,36 @@ def write_parquet(
 
         """
         raise NotImplementedError(f"write_parquet not implemented in {self.__class__}")
+
+    @staticmethod
+    def _check_likely_text_file(resource: URI) -> bool:
+        """Quick sense check of file to see if it looks like text
+        - not 100% full proof, but hopefully enough to weed out most
+        non-text files"""
+        with open_stream(resource, "rb") as fle:
+            start_chunk = fle.read(4096)
+        # check for BOM character - utf-16 can contain NULL bytes
+        if start_chunk.startswith((b"\xff\xfe", b"\xfe\xff")):
+            return True
+        # if null byte in - unlikely text
+        if b"\x00" in start_chunk:
+            return False
+        return True
+
+    def raise_if_not_sensible_file(self, resource: URI, entity_name: str):
+        """Sense check that the file is a text file. Raise error if doesn't
+        appear to be the case."""
+        if not self._check_likely_text_file(resource):
+            raise MessageBearingError(
+                "The submitted file doesn't appear to be text",
+                messages=[
+                    FeedbackMessage(
+                        entity=entity_name,
+                        record=None,
+                        failure_type="submission",
+                        error_location="Whole File",
+                        error_code="MalformedFile",
+                        error_message="The resource doesn't seem to be a valid text file",
+                    )
+                ],
+            )

src/dve/core_engine/backends/base/rules.py

@@ -195,7 +195,7 @@ def evaluate(self, entities, *, config: AbstractStep) -> tuple[Messages, StageSu
                 if success:
                     success = False
                     msg = f"Critical failure in rule {self._step_metadata_to_location(config)}"
-                    self.logger.error(msg)
+                    self.logger.exception(msg)
                 self.logger.error(str(message))
 
         return messages, success
@@ -404,7 +404,7 @@ def apply_sync_filters(
                                 [
                                     CriticalProcessingError(
                                         "Issue occurred while applying filter logic",
-                                        messages=temp_messages,
+                                        messages=[msg.error_message for msg in temp_messages if msg.error_message],
                                     )
                                 ],
                             )
@@ -429,7 +429,7 @@ def apply_sync_filters(
                                 [
                                     CriticalProcessingError(
                                         "Issue occurred while generating FeedbackMessages",
-                                        messages=temp_messages,
+                                        [msg.error_message for msg in temp_messages],
                                     )
                                 ],
                             )
@@ -454,7 +454,7 @@ def apply_sync_filters(
                                 [
                                     CriticalProcessingError(
                                         "Issue occurred while generating FeedbackMessages",
-                                        messages=temp_messages,
+                                        [msg.error_message for msg in temp_messages],
                                     )
                                 ],
                             )
@@ -485,7 +485,7 @@ def apply_sync_filters(
                             [
                                 CriticalProcessingError(
                                     "Issue occurred while filtering error records",
-                                    messages=temp_messages,
+                                    [msg.error_message for msg in temp_messages],
                                 )
                             ],
                         )
@@ -513,7 +513,7 @@ def apply_sync_filters(
                                 [
                                     CriticalProcessingError(
                                         "Issue occurred while generating FeedbackMessages",
-                                        messages=temp_messages,
+                                        [msg.error_message for msg in temp_messages],
                                     )
                                 ],
                             )
@@ -570,7 +570,7 @@ def apply_rules(
                         [
                             CriticalProcessingError(
                                 "Issue occurred while applying pre filter steps",
-                                messages=stage_messages,
+                                [msg.error_message for msg in stage_messages],
                             )
                         ],
                     )
@@ -616,7 +616,7 @@ def apply_rules(
                         [
                             CriticalProcessingError(
                                 "Issue occurred while applying post filter steps",
-                                messages=stage_messages,
+                                [msg.error_message for msg in stage_messages],
                             )
                         ],
                     )

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

@@ -16,6 +16,7 @@
     get_duckdb_type_from_annotation,
 )
 from dve.core_engine.backends.implementations.duckdb.types import SQLType
+from dve.core_engine.backends.readers.utilities import check_csv_header_expected
 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
@@ -24,7 +25,14 @@
 
 @duckdb_write_parquet
 class DuckDBCSVReader(BaseFileReader):
-    """A reader for CSV files"""
+    """A reader for CSV files including the ability to compare the passed model
+    to the file header, if it exists.
+
+    field_check: flag to compare submitted file header to the accompanying pydantic model
+    field_check_error_code: The error code to provide if the file header doesn't contain
+                            the expected fields
+    field_check_error_message: The error message to provide if the file header doesn't contain
+                               the expected fields"""
 
     # TODO - the read_to_relation should include the schema and determine whether to
     # TODO - stringify or not
@@ -35,15 +43,43 @@ def __init__(
         delim: str = ",",
         quotechar: str = '"',
         connection: Optional[DuckDBPyConnection] = None,
+        field_check: bool = False,
+        field_check_error_code: Optional[str] = "ExpectedVsActualFieldMismatch",
+        field_check_error_message: Optional[str] = "The submitted header is missing fields",
         **_,
     ):
         self.header = header
         self.delim = delim
         self.quotechar = quotechar
         self._connection = connection if connection else default_connection
+        self.field_check = field_check
+        self.field_check_error_code = field_check_error_code
+        self.field_check_error_message = field_check_error_message
 
         super().__init__()
 
+    def perform_field_check(
+        self, resource: URI, entity_name: str, expected_schema: type[BaseModel]
+    ):
+        """Check that the header of the CSV aligns with the provided model"""
+        if not self.header:
+            raise ValueError("Cannot perform field check without a CSV header")
+
+        if missing := check_csv_header_expected(resource, expected_schema, self.delim):
+            raise MessageBearingError(
+                "The CSV header doesn't match what is expected",
+                messages=[
+                    FeedbackMessage(
+                        entity=entity_name,
+                        record=None,
+                        failure_type="submission",
+                        error_location="Whole File",
+                        error_code=self.field_check_error_code,
+                        error_message=f"{self.field_check_error_message} - missing fields: {missing}",  # pylint: disable=line-too-long
+                    )
+                ],
+            )
+
     def read_to_py_iterator(
         self, resource: URI, entity_name: EntityName, schema: type[BaseModel]
     ) -> Iterator[dict[str, Any]]:
@@ -58,6 +94,9 @@ def read_to_relation(  # pylint: disable=unused-argument
         if get_content_length(resource) == 0:
             raise EmptyFileError(f"File at {resource} is empty.")
 
+        if self.field_check:
+            self.perform_field_check(resource, entity_name, schema)
+
         reader_options: dict[str, Any] = {
             "header": self.header,
             "delimiter": self.delim,
@@ -89,6 +128,9 @@ def read_to_relation(  # pylint: disable=unused-argument
         if get_content_length(resource) == 0:
             raise EmptyFileError(f"File at {resource} is empty.")
 
+        if self.field_check:
+            self.perform_field_check(resource, entity_name, schema)
+
         reader_options: dict[str, Any] = {
             "has_header": self.header,
             "separator": self.delim,
@@ -132,6 +174,17 @@ class DuckDBCSVRepeatingHeaderReader(PolarsToDuckDBCSVReader):
     | shop1      | clothes    | 2025-01-01 |
     """
 
+    def __init__(
+        self,
+        *args,
+        non_unique_header_error_code: Optional[str] = "NonUniqueHeader",
+        non_unique_header_error_message: Optional[str] = None,
+        **kwargs,
+    ):
+        self._non_unique_header_code = non_unique_header_error_code
+        self._non_unique_header_message = non_unique_header_error_message
+        super().__init__(*args, **kwargs)
+
     @read_function(DuckDBPyRelation)
     def read_to_relation(  # pylint: disable=unused-argument
         self, resource: URI, entity_name: EntityName, schema: type[BaseModel]
@@ -156,10 +209,12 @@ def read_to_relation(  # pylint: disable=unused-argument
                         failure_type="submission",
                         error_message=(
                             f"Found {no_records} distinct combination of header values."
+                            if not self._non_unique_header_message
+                            else self._non_unique_header_message
                         ),
                         error_location=entity_name,
                         category="Bad file",
-                        error_code="NonUniqueHeader",
+                        error_code=self._non_unique_header_code,
                     )
                 ],
             )

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

@@ -0,0 +1,21 @@
+"""General utilities for file readers"""
+
+from typing import Optional
+
+from pydantic import BaseModel
+
+from dve.core_engine.type_hints import URI
+from dve.parser.file_handling.service import open_stream
+
+
+def check_csv_header_expected(
+    resource: URI,
+    expected_schema: type[BaseModel],
+    delimiter: Optional[str] = ",",
+    quote_char: str = '"',
+) -> set[str]:
+    """Check the header of a CSV matches the expected fields"""
+    with open_stream(resource) as fle:
+        header_fields = fle.readline().rstrip().replace(quote_char, "").split(delimiter)
+    expected_fields = expected_schema.__fields__.keys()
+    return set(expected_fields).difference(header_fields)