feat: add facility to add additional checks at end of file transformation stage

Author: stevenhsd

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

@@ -10,6 +10,7 @@
 
 from dve.core_engine.backends.exceptions import MessageBearingError, ReaderLacksEntityTypeSupport
 from dve.core_engine.backends.types import EntityName, EntityType
+from dve.core_engine.configuration.v1 import AllowedAdditionalChecks, _ReaderAdditionalChecksConfig
 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
@@ -109,29 +110,58 @@ def read_to_entity_type(
         entity_name: EntityName,
         schema: type[BaseModel],
         all_model_fields: Optional[set[str]] = None,
-    ) -> EntityType:
+        additional_checks: Optional[
+            dict[AllowedAdditionalChecks, _ReaderAdditionalChecksConfig]
+        ] = None,
+    ):
         """Read to the specified entity type, if supported.
 
         NOTE: Simple types should either be returned as strings (if present) or
         `None`. Format validation, casting, and parsing should be done in the
         data contract.
 
         """
-        if entity_name == Iterator[dict[str, Any]]:
-            return self.read_to_py_iterator(
+        additional_checks = additional_checks or {}
+
+        self.raise_if_not_sensible_file(resource, entity_name)
+
+        if entity_type == Iterator[dict[str, Any]]:
+            entity = self.read_to_py_iterator(
                 resource, entity_name, schema, all_model_fields  # type: ignore
             )
 
-        self.raise_if_not_sensible_file(resource, entity_name)
+        else:
 
-        try:
-            reader_func = self.__read_methods__[entity_type]
-        except KeyError as err:
-            raise ReaderLacksEntityTypeSupport(entity_type=entity_type) from err
+            try:
+                reader_func = self.__read_methods__[entity_type]
+            except KeyError as err:
+                raise ReaderLacksEntityTypeSupport(entity_type=entity_type) from err
 
-        return reader_func(
-            self, resource, entity_name, schema, all_model_fields=all_model_fields  # type: ignore
-        )
+            entity = reader_func(
+                self,
+                resource,
+                entity_name,
+                schema,
+                all_model_fields=all_model_fields,  # type: ignore
+            )
+
+        if config := additional_checks.get("check_empty"):
+            if self.check_entity_empty(entity):
+                raise MessageBearingError(
+                    f"The mandatory entity {entity_name} is empty",
+                    messages=[
+                        FeedbackMessage(
+                            entity=entity_name,
+                            record=None,
+                            failure_type="submission",
+                            error_location=entity_name,
+                            error_code=config.error_code,
+                            error_message=config.error_message,
+                        )
+                    ],
+                )
+
+        return entity
 
     def add_record_index(self, entity: EntityType, **kwargs) -> EntityType:
         """Add a record index to the entity"""
@@ -141,6 +171,10 @@ def drop_record_index(self, entity: EntityType, **kwargs) -> EntityType:
         """Drop a record index to the entity"""
         raise NotImplementedError(f"drop_record_index not implemented in {self.__class__}")
 
+    def check_entity_empty(self, entity: EntityType) -> bool:
+        """Determine if the entity supplied is empty"""
+        raise NotImplementedError(f"check_entity_empty not implemented in {self.__class__}")
+
     def write_parquet(
         self,
         entity: EntityType,

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

@@ -448,8 +448,7 @@ def process_node(node: HierarchyNode):
                                 record=record,  # type: ignore
                                 error_location=location,
                                 error_message=template_object(
-                                    node.missing_parent_id_error_message,
-                                    record
+                                    node.missing_parent_id_error_message, record
                                 ),
                                 failure_type="record",
                                 error_type="record",
@@ -522,10 +521,7 @@ def process_node(node: HierarchyNode) -> bool:
                         entity=node.parent_entity,
                         record=record,  # type: ignore
                         error_location=location,
-                        error_message=template_object(
-                            node.no_valid_records_error_message,
-                            record
-                        ),
+                        error_message=template_object(node.no_valid_records_error_message, record),
                         failure_type="record",
                         error_type="record",
                         error_code=node.no_valid_records_error_code,

src/dve/core_engine/backends/implementations/duckdb/duckdb_helpers.py

@@ -322,6 +322,16 @@ def duckdb_get_entity_count(cls):
     return cls
 
 
+def _duckdb_check_entity_empty(self, entity: DuckDBPyRelation) -> bool:  # pylint: disable=W0613
+    return entity.shape[0] == 0
+
+
+def duckdb_check_entity_empty(cls):
+    """Class decorator to check whether a supplied entity is empty"""
+    cls.check_entity_empty = _duckdb_check_entity_empty
+    return cls
+
+
 def get_all_registered_udfs(connection: DuckDBPyConnection) -> set[str]:
     """Function to supply the names of a registered functions stored in the supplied
     duckdb connection. Creates the temp table used to store registered functions (if not exists).

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

@@ -22,6 +22,7 @@
     UnableToParseCSVError,
 )
 from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import (
+    duckdb_check_entity_empty,
     duckdb_record_index,
     duckdb_write_parquet,
     get_duckdb_type_from_annotation,
@@ -35,6 +36,7 @@
 from dve.parser.file_handling import get_content_length
 
 
+@duckdb_check_entity_empty
 @duckdb_record_index
 @duckdb_write_parquet
 class DuckDBCSVReader(CSVFileReader):

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

@@ -10,6 +10,7 @@
 
 from dve.core_engine.backends.base.reader import BaseFileReader, read_function
 from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import (
+    duckdb_check_entity_empty,
     duckdb_record_index,
     duckdb_write_parquet,
     get_duckdb_type_from_annotation,
@@ -18,6 +19,7 @@
 from dve.core_engine.type_hints import URI, EntityName
 
 
+@duckdb_check_entity_empty
 @duckdb_record_index
 @duckdb_write_parquet
 class DuckDBJSONReader(BaseFileReader):

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

@@ -10,7 +10,10 @@
 
 from dve.core_engine.backends.base.reader import read_function
 from dve.core_engine.backends.exceptions import MessageBearingError
-from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import duckdb_write_parquet
+from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import (
+    duckdb_check_entity_empty,
+    duckdb_write_parquet,
+)
 from dve.core_engine.backends.readers.xml import XMLStreamReader
 from dve.core_engine.backends.utilities import (
     get_polars_type_from_annotation,
@@ -20,6 +23,7 @@
 from dve.core_engine.type_hints import URI
 
 
+@duckdb_check_entity_empty
 @polars_record_index
 @duckdb_write_parquet
 class DuckDBXMLStreamReader(XMLStreamReader):

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

@@ -12,6 +12,7 @@
 from dve.core_engine.backends.exceptions import EmptyFileError
 from dve.core_engine.backends.implementations.spark.spark_helpers import (
     get_type_from_annotation,
+    spark_check_entity_empty,
     spark_record_index,
     spark_write_parquet,
 )
@@ -20,6 +21,7 @@
 from dve.parser.file_handling import get_content_length
 
 
+@spark_check_entity_empty
 @spark_record_index
 @spark_write_parquet
 class SparkCSVReader(CSVFileReader):

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

@@ -11,13 +11,15 @@
 from dve.core_engine.backends.exceptions import EmptyFileError
 from dve.core_engine.backends.implementations.spark.spark_helpers import (
     get_type_from_annotation,
+    spark_check_entity_empty,
     spark_record_index,
     spark_write_parquet,
 )
 from dve.core_engine.type_hints import URI, EntityName
 from dve.parser.file_handling import get_content_length
 
 
+@spark_check_entity_empty
 @spark_record_index
 @spark_write_parquet
 class SparkJSONReader(BaseFileReader):

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

@@ -17,6 +17,7 @@
 from dve.core_engine.backends.implementations.spark.spark_helpers import (
     df_is_empty,
     get_type_from_annotation,
+    spark_check_entity_empty,
     spark_record_index,
     spark_write_parquet,
 )
@@ -29,6 +30,7 @@
 """The mode to use when parsing XML files with Spark."""
 
 
+@spark_check_entity_empty
 @spark_record_index
 @spark_write_parquet
 class SparkXMLStreamReader(XMLStreamReader):
@@ -55,6 +57,7 @@ def read_to_dataframe(
         )
 
 
+@spark_check_entity_empty
 @spark_record_index
 @spark_write_parquet
 class SparkXMLReader(BasicXMLFileReader):  # pylint: disable=too-many-instance-attributes

src/dve/core_engine/backends/implementations/spark/spark_helpers.py

@@ -456,6 +456,16 @@ def spark_get_entity_count(cls):
     return cls
 
 
+def _spark_check_entity_empty(self, entity: DataFrame) -> bool:  # pylint: disable=W0613
+    return entity.count() == 0
+
+
+def spark_check_entity_empty(cls):
+    """Class decorator to check whether a supplied entity is empty"""
+    cls.check_entity_empty = _spark_check_entity_empty
+    return cls
+
+
 def get_all_registered_udfs(spark: SparkSession) -> set[str]:
     """Function to supply the names of a registered functions stored in the supplied
     spark session.