feat: Added new option to check csv headers in duckdb csv readers

Author: stevenhsd

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

@@ -16,10 +16,11 @@
     get_duckdb_type_from_annotation,
 )
 from dve.core_engine.backends.implementations.duckdb.types import SQLType
+from dve.core_engine.backends.implementations.duckdb.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
-from dve.parser.file_handling import get_content_length
+from dve.parser.file_handling import get_content_length, open_stream
 
 
 @duckdb_write_parquet
@@ -35,15 +36,46 @@ 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 does not match what is expected",
         **_,
     ):
         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]
+    ):
+        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,
+                        failure_type="submission",
+                        error_location="Whole File",
+                        error_code=self.field_check_error_code,
+                        error_message=self.field_check_error_message,
+                        value=f"Missing fields: {missing}",
+                    )
+                ],
+            )
+
     def read_to_py_iterator(
         self, resource: URI, entity_name: EntityName, schema: type[BaseModel]
     ) -> Iterator[dict[str, Any]]:
@@ -58,6 +90,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 +124,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 +170,12 @@ class DuckDBCSVRepeatingHeaderReader(PolarsToDuckDBCSVReader):
     | shop1      | clothes    | 2025-01-01 |
     """
 
+    def __init__(
+        self, non_unique_header_error_code: Optional[str] = "NonUniqueHeader", *args, **kwargs
+    ):
+        self._non_unique_header_code = non_unique_header_error_code
+        super().__init__(*args, **kwargs)
+
     @read_function(DuckDBPyRelation)
     def read_to_relation(  # pylint: disable=unused-argument
         self, resource: URI, entity_name: EntityName, schema: type[BaseModel]
@@ -159,7 +203,7 @@ def read_to_relation(  # pylint: disable=unused-argument
                         ),
                         error_location=entity_name,
                         category="Bad file",
-                        error_code="NonUniqueHeader",
+                        error_code=self._non_unique_header_code,
                     )
                 ],
             )

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

@@ -1,8 +1,15 @@
 """Utility objects for use with duckdb backend"""
 
 import itertools
+from typing import Optional
+
+from pydantic import BaseModel
 
 from dve.core_engine.backends.base.utilities import _split_multiexpr_string
+from dve.core_engine.backends.exceptions import MessageBearingError
+from dve.core_engine.message import FeedbackMessage
+from dve.core_engine.type_hints import URI
+from dve.parser.file_handling import open_stream
 
 
 def parse_multiple_expressions(expressions) -> list[str]:
@@ -39,3 +46,15 @@ def multiexpr_string_to_columns(expressions: str) -> list[str]:
     """
     expression_list = _split_multiexpr_string(expressions)
     return expr_array_to_columns(expression_list)
+
+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().replace(quote_char,"").split(delimiter)
+    expected_fields = expected_schema.__fields__.keys()
+    return set(expected_fields).difference(header_fields)
+    

tests/test_core_engine/test_backends/test_implementations/test_duckdb/test_ddb_utils.py

@@ -1,9 +1,14 @@
-from typing import Dict, List
+import tempfile
+import datetime as dt
+from pathlib import Path
+from uuid import uuid4
+from pydantic import BaseModel, create_model
 import pytest
 
 from dve.core_engine.backends.implementations.duckdb.utilities import (
     expr_mapping_to_columns,
     expr_array_to_columns,
+    check_csv_header_expected,
 )
 
 
@@ -16,7 +21,7 @@
         ),
     ],
 )
-def test_expr_mapping_to_columns(expressions: Dict[str, str], expected: list[str]):
+def test_expr_mapping_to_columns(expressions: dict[str, str], expected: list[str]):
     observed = expr_mapping_to_columns(expressions)
     assert observed == expected
 
@@ -51,6 +56,47 @@ def test_expr_mapping_to_columns(expressions: Dict[str, str], expected: list[str
         ),
     ],
 )
-def test_expr_array_to_columns(expressions: Dict[str, str], expected: list[str]):
+def test_expr_array_to_columns(expressions: dict[str, str], expected: list[str]):
     observed = expr_array_to_columns(expressions)
     assert observed == expected
+
+
+@pytest.mark.parametrize(
+    ["header_row", "delim", "schema", "expected"],
+    [
+        (
+            "field1,field2,field3",
+            ",",
+            {"field1": (str, ...), "field2": (int, ...), "field3": (float, 1.2)},
+            set(),
+        ),
+        (
+            "field2,field3,field1",
+            ",",
+            {"field1": (str, ...), "field2": (int, ...), "field3": (float, 1.2)},
+            set(),
+        ),
+        (
+            "str_field|int_field|date_field|",
+            ",",
+            {"str_field": (str, ...), "int_field": (int, ...), "date_field": (dt.date, dt.date.today())},
+            {"str_field","int_field","date_field"},
+        ),
+        (
+            '"str_field"|"int_field"|"date_field"',
+            "|",
+            {"str_field": (str, ...), "int_field": (int, ...), "date_field": (dt.date, dt.date.today())},
+            set(),
+        ),
+        
+    ],
+)
+def test_check_csv_header_expected(
+    header_row: str, delim: str, schema: type[BaseModel], expected: set[str]
+):
+    mdl = create_model("TestModel", **schema)
+    with tempfile.TemporaryDirectory() as tmpdir:
+        fle = Path(tmpdir).joinpath(f"test_file_{uuid4().hex}.csv")
+        fle.open("w+").write(header_row)
+        res = check_csv_header_expected(fle.as_posix(), mdl, delim)
+    assert res == expected