refactor: add CriticalFeedbackMessage which exits the pipeline asap

number of feedback messages to reduce and removal of specific error code+messages for certain readers to become generic ft_error_code/message

Author: georgeRobertson

docs/advanced_guidance/json_schemas/contract/components/reader_constraints/global_csv_reader_args.schema.json

@@ -4,6 +4,11 @@
     "title": "Keyword Arguments used across all CSV readers",
     "description": "Arguments present in all CSV readers available.",
     "type": "object",
+    "anyOf": [
+        {
+            "$ref": "global_reader_args.schema.json"
+        }
+    ],
     "properties": {
         "field_check": {
             "type": "string",
@@ -13,14 +18,6 @@
                 "false"
             ]
         },
-        "field_check_error_code": {
-            "type": "string",
-            "description": "Allows you to customise the error code generated in the error report when the field check fails."
-        },
-        "field_check_error_message": {
-            "type": "string",
-            "description": "Allows you to customise the error message generated in the error report when the field check fails."
-        },
         "null_empty_strings": {
             "type": "boolean",
             "description": "Converts empty string values into 'Null' values."

docs/advanced_guidance/json_schemas/contract/components/reader_constraints/global_reader_args.schema.json

@@ -0,0 +1,19 @@
+{
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "$id": "data-ingest:contract/components/reader_constraints/global_reader_args.schema.json",
+    "title": "Keyword Arguments used across all readers",
+    "description": "Arguments present in all readers available.",
+    "type": "object",
+    "properties": {
+        "ft_error_code": {
+            "type": "string",
+            "description": "Error code to raise when a reader is unable to read the contents of a file.",
+            "minLength": 1
+        },
+        "ft_error_message": {
+            "type": "string",
+            "description": "Error message to raise when a reader is unable to read the contents of a file.",
+            "minLength": 1
+        }
+    }
+}

docs/advanced_guidance/json_schemas/contract/components/reader_constraints/global_xml_reader_args.schema.json

@@ -4,6 +4,11 @@
     "title": "Keyword Arguments used across all XML readers",
     "description": "Arguments present in all XML readers.",
     "type": "object",
+    "anyOf": [
+        {
+            "$ref": "global_reader_args.schema.json"
+        }
+    ],
     "properties": {
         "record_tag": {
             "type": "string",
@@ -40,14 +45,6 @@
             "type": "string",
             "description": "The Relative path (to the dischema) for the XSD document."
         },
-        "xsd_error_code": {
-            "type": "string",
-            "description": "Allows you to customise the error code generated in the error report when the XSD check fails."
-        },
-        "xsd_error_message": {
-            "type": "string",
-            "description": "Allows you to customise the error message generated in the error report when the XSD check fails."
-        },
         "rules_location": {
             "type": "string",
             "description": "Allows you to prefix the directory which contains the XSD document."

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

@@ -73,6 +73,10 @@ class BaseFileReader(ABC):
     decorated with the '@read_function' decorator, and is used in `read_entity_type`.
 
     """
+    ft_error_code: Optional[str] = "MalformedFile"
+    """Default error code for when a file/submission cannot be parsed succesfully."""
+    ft_error_message: Optional[str] = "The resource doesn't seem to be a valid text file"
+    """Default error message for when a file/submission cannot be parsed succesfully."""
 
     def __init_subclass__(cls, *_, **__) -> None:
         """When this class is subclassed, create and populate the `__read_methods__`
@@ -212,7 +216,11 @@ def _check_likely_text_file(resource: URI) -> bool:
             return False
         return True
 
-    def raise_if_not_sensible_file(self, resource: URI, entity_name: str):
+    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):
@@ -223,7 +231,7 @@ def raise_if_not_sensible_file(self, resource: URI, entity_name: str):
                     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",
+                    error_code=self.ft_error_code,
+                    error_message=self.ft_error_message,
                 ),
             )

src/dve/core_engine/backends/exceptions.py

@@ -44,27 +44,25 @@ def __init__(self, *args: object, message: FeedbackMessage) -> None:
         self.message = message
         """The message to be returned as part of the error."""
 
-class UnableToParseCSVError(MessageBearingError):
+class UnableToParseCSVError(CriticalMessageBearingError):
     """An error raised when unable to parse a CSV file"""
 
     def __init__(
-        self, entity_name: str, field_check_error_message: str, field_check_error_code: str
+        self,
+        entity_name: Optional[str],
+        error_message: Optional[str],
+        error_code: Optional[str],
     ):
         super().__init__(
-            messages=[
-                FeedbackMessage(
-                    entity="csv_structure",
-                    record={
-                        entity_name: "Unable to parse file. Please check the structure of the file."
-                    },
-                    failure_type="submission",
-                    is_informational=False,
-                    error_type="csv read",
-                    error_location=entity_name,
-                    error_message=field_check_error_message,
-                    error_code=field_check_error_code,
-                )
-            ]
+            message=FeedbackMessage(
+                entity=entity_name,
+                record=None,
+                failure_type="submission",
+                is_informational=False,
+                error_type="csv read",
+                error_message=error_message or "Unable to parse the CSV file. Please check the structure of your CSV.",  # pylint: disable=C0301
+                error_code=error_code or "MalformedCSV",
+            )
         )
 
 

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

@@ -26,11 +26,12 @@
     duckdb_record_index,
     duckdb_write_parquet,
     get_duckdb_type_from_annotation,
+    relation_is_empty,
 )
 from dve.core_engine.backends.implementations.duckdb.types import SQLType
 from dve.core_engine.backends.readers.csv import CSVFileReader
 from dve.core_engine.backends.utilities import get_polars_type_from_annotation, polars_record_index
-from dve.core_engine.constants import RECORD_INDEX_COLUMN_NAME
+from dve.core_engine.constants import PRE_VALIDATION_ENTITY, RECORD_INDEX_COLUMN_NAME
 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
@@ -44,10 +45,7 @@ class DuckDBCSVReader(CSVFileReader):
     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
@@ -59,8 +57,8 @@ def __init__(
         quotechar: str = '"',
         connection: Optional[DuckDBPyConnection] = None,
         field_check: bool = False,
-        field_check_error_code: str = "ExpectedVsActualFieldMismatch",
-        field_check_error_message: str = "The submitted header is missing fields",
+        ft_error_code: str = "ExpectedVsActualFieldMismatch",
+        ft_error_message: str = "The submitted header is missing fields",
         null_empty_strings: bool = False,
         **_,
     ):
@@ -72,8 +70,8 @@ def __init__(
             delimiter=delim,
             quote_char=quotechar,
             field_check=field_check,
-            field_check_error_code=field_check_error_code,
-            field_check_error_message=field_check_error_message,
+            ft_error_code=ft_error_code,
+            ft_error_message=ft_error_message,
         )
 
     def read_to_py_iterator(
@@ -124,8 +122,8 @@ def read_to_relation(  # pylint: disable=unused-argument
         except InvalidInputException as exc:
             raise UnableToParseCSVError(
                 entity_name="csv_structure",
-                field_check_error_message=self.field_check_error_message,
-                field_check_error_code=self.field_check_error_code,
+                error_code=self.ft_error_code,
+                error_message=self.ft_error_message or "Unable to parse CSV file. Structure is likely malformed.",  # pylint: disable=C0301
             ) from exc
 
         if self.null_empty_strings:
@@ -184,8 +182,8 @@ def read_to_relation(  # pylint: disable=unused-argument
         except pl.exceptions.PolarsError as exc:
             raise UnableToParseCSVError(
                 entity_name="csv_structure",
-                field_check_error_message=self.field_check_error_message,
-                field_check_error_code=self.field_check_error_code,
+                error_code=self.ft_error_code,
+                error_message=self.ft_error_message or "Unable to parse CSV file. Structure is likely malformed.",  # pylint: disable=C0301
             ) from exc
 
         if self.null_empty_strings:
@@ -198,11 +196,11 @@ def read_to_relation(  # pylint: disable=unused-argument
 
         entity = self._connection.sql("SELECT * FROM df")
 
-        if entity.pl().shape[0] == 0:
+        if relation_is_empty(entity):
             raise UnableToParseCSVError(
                 entity_name="csv_structure",
-                field_check_error_message=self.field_check_error_message,
-                field_check_error_code=self.field_check_error_code,
+                error_code=self.ft_error_code,
+                error_message=self.ft_error_message or "Found zero records after loading CSV. File is likely malformed.",  # pylint: disable=C0301
             )
 
         return entity
@@ -273,7 +271,7 @@ def read_to_relation(  # pylint: disable=unused-argument
                 messages=[
                     FeedbackMessage(
                         record={entity_name: differing_values},
-                        entity="Pre-validation",
+                        entity=PRE_VALIDATION_ENTITY,
                         failure_type="submission",
                         error_message=(
                             f"Found {no_records} distinct combination of header values."

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

@@ -40,8 +40,8 @@ def __init__(
         null_empty_strings: bool = False,
         spark_session: Optional[SparkSession] = None,
         field_check: bool = False,
-        field_check_error_code: str = "ExpectedVsActualFieldMismatch",
-        field_check_error_message: str = "The submitted header is missing fields",
+        ft_error_code: str = "ExpectedVsActualFieldMismatch",
+        ft_error_message: str = "The submitted header is missing fields",
         **_,
     ) -> None:
 
@@ -56,8 +56,8 @@ def __init__(
             quote_char=quote_char,
             header=header,
             field_check=field_check,
-            field_check_error_code=field_check_error_code,
-            field_check_error_message=field_check_error_message,
+            ft_error_code=ft_error_code,
+            ft_error_message=ft_error_message,
         )
 
     def read_to_py_iterator(

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

@@ -79,8 +79,8 @@ def __init__(
         namespace=None,
         trim_cells=True,
         xsd_location: Optional[URI] = None,
-        xsd_error_code: Optional[str] = None,
-        xsd_error_message: Optional[str] = None,
+        ft_error_code: Optional[str] = None,
+        ft_error_message: Optional[str] = None,
         rules_location: Optional[URI] = None,
         **_,
     ) -> None:
@@ -92,8 +92,8 @@ def __init__(
             null_values=null_values,
             sanitise_multiline=sanitise_multiline,
             xsd_location=xsd_location,
-            xsd_error_code=xsd_error_code,
-            xsd_error_message=xsd_error_message,
+            ft_error_code=ft_error_code,
+            ft_error_message=ft_error_message,
             rules_location=rules_location,
         )
 

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

@@ -42,8 +42,8 @@ def __init__(
         null_values: Collection[str] = frozenset({"NULL", "null", ""}),
         encoding: str = "utf-8-sig",
         field_check: bool = False,
-        field_check_error_code: str = "CSVFieldMismatch",
-        field_check_error_message: str = "The submitted header is invalid",
+        ft_error_code: Optional[str] = "MalformedCSVFile",
+        ft_error_message: Optional[str] = None,
         **_,
     ):
         """Init function for the base CSV reader.
@@ -89,10 +89,8 @@ def __init__(
         """Encoding of the CSV file."""
         self.field_check = field_check
         """Whether to check the fields are correct in the supplied header or not"""
-        self.field_check_error_code = field_check_error_code
-        """Error code to raise when fields are missing or unexpected"""
-        self.field_check_error_message = field_check_error_message
-        """Error message to raise when fields are missing or unexpected"""
+        self.ft_error_code = ft_error_code
+        self.ft_error_message = ft_error_message
 
     def _get_reader_args(self) -> dict[str, Any]:
         reader_args: dict[str, Any] = {
@@ -218,8 +216,8 @@ def perform_field_check(
             entity_name,
             expected_schema,
             all_model_fields,
-            self.field_check_error_code,
-            self.field_check_error_message,
+            self.ft_error_code or "CSVFieldMismatch",
+            self.ft_error_message or "The submitted header is invalid",
             self.delimiter,
             self.quote_char,
         )

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

@@ -5,7 +5,8 @@
 
 from pydantic import BaseModel
 
-from dve.core_engine.backends.exceptions import MessageBearingError
+from dve.core_engine.backends.exceptions import CriticalMessageBearingError
+from dve.core_engine.constants import PRE_VALIDATION_ENTITY
 from dve.core_engine.message import FeedbackMessage
 from dve.core_engine.type_hints import URI, EntityName
 from dve.parser.file_handling.service import open_stream
@@ -55,19 +56,17 @@ def raise_message_bearing_error_on_header_differences(
         record_details_additional = (
             f"additional fields: {', '.join(sorted(additional))};" if additional else ""
         )  # pylint: disable=C0301
-        raise MessageBearingError(
+        raise CriticalMessageBearingError(
             "The CSV header doesn't match what is expected",
-            messages=[
-                FeedbackMessage(
-                    entity="Pre-validation",
-                    record={entity_name: f"{record_details_missing}{record_details_additional}"},
-                    failure_type="submission",
-                    error_location=entity_name,
-                    reporting_field="csv_header",
-                    error_code=field_check_error_code,
-                    error_message=field_check_error_message,
-                )
-            ],
+            message=FeedbackMessage(
+                entity=PRE_VALIDATION_ENTITY,
+                record={entity_name: f"{record_details_missing}{record_details_additional}"},
+                failure_type="submission",
+                error_location=entity_name,
+                reporting_field="csv_header",
+                error_code=field_check_error_code,
+                error_message=field_check_error_message,
+            )
         )