fix: add all available model fields to ensure that multiple models don't trigger additional fields check for csv

restructured the reader tests as well

Author: georgeRobertson

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

@@ -90,6 +90,7 @@ def read_to_py_iterator(
         resource: URI,
         entity_name: EntityName,
         schema: type[BaseModel],
+        all_model_fields: Optional[set[str]] = None,
     ) -> Iterator[dict[str, Any]]:
         """Iterate through the contents of the resource, yielding dicts
         representing each record.
@@ -107,6 +108,7 @@ def read_to_entity_type(
         resource: URI,
         entity_name: EntityName,
         schema: type[BaseModel],
+        all_model_fields: Optional[set[str]] = None,
     ) -> EntityType:
         """Read to the specified entity type, if supported.
 
@@ -116,7 +118,12 @@ def read_to_entity_type(
 
         """
         if entity_name == Iterator[dict[str, Any]]:
-            return self.read_to_py_iterator(resource, entity_name, schema)  # type: ignore
+            return self.read_to_py_iterator(
+                resource,
+                entity_name,
+                schema, # type: ignore
+                all_model_fields
+            )
 
         self.raise_if_not_sensible_file(resource, entity_name)
 
@@ -125,7 +132,7 @@ def read_to_entity_type(
         except KeyError as err:
             raise ReaderLacksEntityTypeSupport(entity_type=entity_type) from err
 
-        return reader_func(self, resource, entity_name, schema)
+        return reader_func(self, resource, entity_name, schema, all_model_fields=all_model_fields)
 
     def add_record_index(self, entity: EntityType, **kwargs) -> EntityType:
         """Add a record index to the entity"""

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

@@ -9,17 +9,15 @@
 from duckdb import DuckDBPyConnection, DuckDBPyRelation, StarExpression, read_csv
 from pydantic import BaseModel
 
-from dve.core_engine.backends.base.reader import BaseFileReader, read_function
+from dve.core_engine.backends.base.reader import read_function
 from dve.core_engine.backends.exceptions import EmptyFileError, MessageBearingError
 from dve.core_engine.backends.implementations.duckdb.duckdb_helpers import (
     duckdb_record_index,
     duckdb_write_parquet,
     get_duckdb_type_from_annotation,
 )
 from dve.core_engine.backends.implementations.duckdb.types import SQLType
-from dve.core_engine.backends.readers.utilities import (
-    raise_message_bearing_error_on_header_differences,
-)
+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.message import FeedbackMessage
@@ -29,7 +27,7 @@
 
 @duckdb_record_index
 @duckdb_write_parquet
-class DuckDBCSVReader(BaseFileReader):
+class DuckDBCSVReader(CSVFileReader):
     """A reader for CSV files including the ability to compare the passed model
     to the file header, if it exists.
 
@@ -54,55 +52,52 @@ def __init__(
         null_empty_strings: bool = False,
         **_,
     ):
-        self.header = header
-        self.delim = delim
-        self.quotechar = quotechar
         self._connection = connection if connection else ddb.connect(":memory:")
-        self.field_check = field_check
-        self.field_check_error_code = field_check_error_code
-        self.field_check_error_message = field_check_error_message
         self.null_empty_strings = null_empty_strings
 
-        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")
-
-        raise_message_bearing_error_on_header_differences(
-            resource,
-            entity_name,
-            expected_schema,
-            self.field_check_error_code,
-            self.field_check_error_message,
-            self.delim,
-            self.quotechar,
+        super().__init__(
+            header=header,
+            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
         )
 
     def read_to_py_iterator(
-        self, resource: URI, entity_name: EntityName, schema: type[BaseModel]
+        self,
+        resource: URI,
+        entity_name: EntityName,
+        schema: type[BaseModel],
+        all_model_fields: Optional[set[str]] = None,
     ) -> Iterator[dict[str, Any]]:
         """Creates an iterable object of rows as dictionaries"""
-        yield from self.read_to_relation(resource, entity_name, schema).pl().iter_rows(named=True)
+        yield from self.read_to_relation(
+            resource,
+            entity_name,
+            schema,
+            all_model_fields,
+        ).pl().iter_rows(named=True)
 
     @read_function(DuckDBPyRelation)
     def read_to_relation(  # pylint: disable=unused-argument
-        self, resource: URI, entity_name: EntityName, schema: type[BaseModel]
+        self,
+        resource: URI,
+        entity_name: EntityName,
+        schema: type[BaseModel],
+        all_model_fields: Optional[set[str]] = None,
     ) -> DuckDBPyRelation:
         """Returns a relation object from the source csv"""
         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)
+            self.perform_field_check(resource, entity_name, schema, all_model_fields)
 
         reader_options: dict[str, Any] = {
             "header": self.header,
-            "delimiter": self.delim,
-            "quotechar": self.quotechar,
+            "delimiter": self.delimiter,
+            "quotechar": self.quote_char,
         }
 
         ddb_schema: dict[str, SQLType] = {
@@ -134,19 +129,23 @@ class PolarsToDuckDBCSVReader(DuckDBCSVReader):
 
     @read_function(DuckDBPyRelation)
     def read_to_relation(  # pylint: disable=unused-argument
-        self, resource: URI, entity_name: EntityName, schema: type[BaseModel]
+        self,
+        resource: URI,
+        entity_name: EntityName,
+        schema: type[BaseModel],
+        all_model_fields: Optional[set[str]] = None,
     ) -> DuckDBPyRelation:
         """Returns a relation object from the source csv"""
         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)
+            self.perform_field_check(resource, entity_name, schema, all_model_fields)
 
         reader_options: dict[str, Any] = {
             "has_header": self.header,
-            "separator": self.delim,
-            "quote_char": self.quotechar,
+            "separator": self.delimiter,
+            "quote_char": self.quote_char,
         }
 
         polars_types = {
@@ -212,10 +211,17 @@ def __init__(
 
     @read_function(DuckDBPyRelation)
     def read_to_relation(  # pylint: disable=unused-argument
-        self, resource: URI, entity_name: EntityName, schema: type[BaseModel]
+        self,
+        resource: URI,
+        entity_name: EntityName,
+        schema: type[BaseModel],
+        all_model_fields: Optional[set[str]] = None,
     ) -> DuckDBPyRelation:
         entity: DuckDBPyRelation = super().read_to_relation(
-            resource=resource, entity_name=entity_name, schema=schema
+            resource=resource,
+            entity_name=entity_name,
+            schema=schema,
+            all_model_fields=all_model_fields
         )
         entity = entity.select(StarExpression(exclude=[RECORD_INDEX_COLUMN_NAME])).distinct()
         no_records = entity.shape[0]

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

@@ -36,14 +36,22 @@ def __init__(
         super().__init__()
 
     def read_to_py_iterator(
-        self, resource: URI, entity_name: EntityName, schema: type[BaseModel]
+        self,
+        resource: URI,
+        entity_name: EntityName,
+        schema: type[BaseModel],
+        all_model_fields: Optional[set[str]] = None,
     ) -> Iterator[dict[str, Any]]:
         """Creates an iterable object of rows as dictionaries"""
         return self.read_to_relation(resource, entity_name, schema).pl().iter_rows(named=True)
 
     @read_function(DuckDBPyRelation)
     def read_to_relation(  # pylint: disable=unused-argument
-        self, resource: URI, entity_name: EntityName, schema: type[BaseModel]
+        self,
+        resource: URI,
+        entity_name: EntityName,
+        schema: type[BaseModel],
+        **_,
     ) -> DuckDBPyRelation:
         """Returns a relation object from the source json"""
 

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

@@ -30,7 +30,13 @@ def __init__(self, *, connection: Optional[DuckDBPyConnection] = None, **kwargs)
         super().__init__(**kwargs)
 
     @read_function(DuckDBPyRelation)
-    def read_to_relation(self, resource: URI, entity_name: str, schema: type[BaseModel]):
+    def read_to_relation(
+        self,
+        resource: URI,
+        entity_name: str,
+        schema: type[BaseModel],
+        **_,
+    ):
         """Returns a relation object from the source xml"""
         if self.xsd_location:
             msg = self._run_xmllint(file_uri=resource)

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

@@ -8,23 +8,21 @@
 from pyspark.sql import DataFrame, SparkSession
 from pyspark.sql.types import StructType
 
-from dve.core_engine.backends.base.reader import BaseFileReader, read_function
+from dve.core_engine.backends.base.reader import read_function
+from dve.core_engine.backends.readers.csv import CSVFileReader
 from dve.core_engine.backends.exceptions import EmptyFileError
 from dve.core_engine.backends.implementations.spark.spark_helpers import (
     get_type_from_annotation,
     spark_record_index,
     spark_write_parquet,
 )
-from dve.core_engine.backends.readers.utilities import (
-    raise_message_bearing_error_on_header_differences,
-)
 from dve.core_engine.type_hints import URI, EntityName
 from dve.parser.file_handling import get_content_length
 
 
 @spark_record_index
 @spark_write_parquet
-class SparkCSVReader(BaseFileReader):
+class SparkCSVReader(CSVFileReader):
     """A Spark reader for CSV files."""
 
     # pylint: disable=R0902
@@ -45,41 +43,29 @@ def __init__(
         **_,
     ) -> None:
 
-        self.delimiter = delimiter
-        self.escape_char = escape_char
-        self.encoding = encoding
-        self.quote_char = quote_char
-        self.header = header
         self.multi_line = multi_line
         self.null_empty_strings = null_empty_strings
         self.spark_session = spark_session if spark_session else SparkSession.builder.getOrCreate()  # type: ignore  # pylint: disable=C0301
-        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")
 
-        raise_message_bearing_error_on_header_differences(
-            resource,
-            entity_name,
-            expected_schema,
-            self.field_check_error_code,
-            self.field_check_error_message,
-            self.delimiter,
-            self.quote_char,
+        super().__init__(
+            delimiter=delimiter,
+            escape_char=escape_char,
+            encoding=encoding,
+            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,
         )
 
     def read_to_py_iterator(
-        self, resource: URI, entity_name: EntityName, schema: type[BaseModel]
+        self,
+        resource: URI,
+        entity_name: EntityName,
+        schema: type[BaseModel],
+        all_model_fields: Optional[set[str]] = None,
     ) -> Iterator[dict[URI, Any]]:
-        df = self.read_to_dataframe(resource, entity_name, schema)
+        df = self.read_to_dataframe(resource, entity_name, schema, all_model_fields)
         yield from (record.asDict(True) for record in df.toLocalIterator())
 
     @read_function(DataFrame)
@@ -88,13 +74,14 @@ def read_to_dataframe(
         resource: URI,
         entity_name: EntityName,  # pylint: disable=unused-argument
         schema: type[BaseModel],
+        all_model_fields: Optional[set[str]] = None,
     ) -> DataFrame:
         """Read a CSV file directly to a Spark DataFrame."""
         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)
+            self.perform_field_check(resource, entity_name, schema, all_model_fields)
 
         spark_schema: StructType = get_type_from_annotation(schema)
         kwargs = {

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

@@ -39,7 +39,11 @@ def __init__(
         super().__init__()
 
     def read_to_py_iterator(
-        self, resource: URI, entity_name: EntityName, schema: type[BaseModel]
+        self,
+        resource: URI,
+        entity_name: EntityName,
+        schema: type[BaseModel],
+        all_model_fields: Optional[set[str]] = None,
     ) -> Iterator[dict[URI, Any]]:
         df = self.read_to_dataframe(resource, entity_name, schema)
         yield from (record.asDict(True) for record in df.toLocalIterator())
@@ -50,6 +54,7 @@ def read_to_dataframe(
         resource: URI,
         entity_name: EntityName,  # pylint: disable=unused-argument
         schema: type[BaseModel],
+        **_,
     ) -> DataFrame:
         """Read a JSON file directly to a Spark DataFrame."""
         if get_content_length(resource) == 0:

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

@@ -104,7 +104,11 @@ def __init__(
         self.namespace = namespace
 
     def read_to_py_iterator(
-        self, resource: URI, entity_name: EntityName, schema: type[BaseModel]
+        self,
+        resource: URI,
+        entity_name: EntityName,
+        schema: type[BaseModel],
+        all_model_fields: Optional[set[str]] = None,
     ) -> Iterator[dict[URI, Any]]:
         df = self.read_to_dataframe(resource, entity_name, schema)
         yield from (record.asDict(True) for record in df.toLocalIterator())
@@ -115,6 +119,7 @@ def read_to_dataframe(
         resource: URI,
         entity_name: EntityName,  # pylint: disable=unused-argument
         schema: type[BaseModel],
+        **_,
     ) -> DataFrame:
         """Read an XML file directly to a Spark DataFrame using the Databricks
         XML reader package.