fix: fix issues and implementation logic for xsd validation

Author: georgeRobertson

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

@@ -30,10 +30,12 @@ class DuckDBCSVReader(BaseFileReader):
     # TODO - stringify or not
     def __init__(
         self,
+        *,
         header: bool = True,
         delim: str = ",",
         quotechar: str = '"',
         connection: Optional[DuckDBPyConnection] = None,
+        **_,
     ):
         self.header = header
         self.delim = delim

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

@@ -20,7 +20,12 @@
 class DuckDBJSONReader(BaseFileReader):
     """A reader for JSON files"""
 
-    def __init__(self, json_format: Optional[str] = "array"):
+    def __init__(
+        self,
+        *,
+        json_format: Optional[str] = "array",
+        **_,
+    ):
         self._json_format = json_format
 
         super().__init__()

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

@@ -8,21 +8,18 @@
 from pydantic import BaseModel
 
 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.readers.xml import XMLStreamReader
 from dve.core_engine.backends.utilities import get_polars_type_from_annotation, stringify_model
 from dve.core_engine.type_hints import URI
-from dve.parser.file_handling.service import get_parent
-from dve.pipeline.utils import dump_errors
 
 
 @duckdb_write_parquet
 class DuckDBXMLStreamReader(XMLStreamReader):
     """A reader for XML files"""
 
-    def __init__(self,
-                 ddb_connection: Optional[DuckDBPyConnection] = None,
-                 **kwargs):
+    def __init__(self, *, ddb_connection: Optional[DuckDBPyConnection] = None, **kwargs):
         self.ddb_connection = ddb_connection if ddb_connection else default_connection
         super().__init__(**kwargs)
 
@@ -32,14 +29,12 @@ def read_to_relation(self, resource: URI, entity_name: str, schema: type[BaseMod
         if self.xsd_location:
             msg = self._run_xmllint(file_uri=resource)
             if msg:
-                working_folder = get_parent(resource)
-                dump_errors(
-                    working_folder=working_folder,
-                    step_name="file_transformation",
-                    messages=[msg]
-                    )
-
-        polars_schema: Dict[str, pl.DataType] = {  # type: ignore
+                raise MessageBearingError(
+                    "Submitted file failed XSD validation.",
+                    messages=[msg],
+                )
+
+        polars_schema: dict[str, pl.DataType] = {  # type: ignore
             fld.name: get_polars_type_from_annotation(fld.annotation)
             for fld in stringify_model(schema).__fields__.values()
         }

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

@@ -31,6 +31,7 @@ def __init__(
         multi_line: bool = False,
         encoding: str = "utf-8-sig",
         spark_session: Optional[SparkSession] = None,
+        **_,
     ) -> None:
 
         self.delimiter = delimiter

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

@@ -27,6 +27,7 @@ def __init__(
         encoding: Optional[str] = "utf-8",
         multi_line: Optional[bool] = True,
         spark_session: Optional[SparkSession] = None,
+        **_,
     ) -> None:
 
         self.encoding = encoding

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

@@ -12,19 +12,18 @@
 from pyspark.sql.utils import AnalysisException
 from typing_extensions import Literal
 
-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
 from dve.core_engine.backends.implementations.spark.spark_helpers import (
     df_is_empty,
     get_type_from_annotation,
     spark_write_parquet,
 )
 from dve.core_engine.backends.readers.xml import BasicXMLFileReader, XMLStreamReader
-from dve.core_engine.backends.readers.xml_linting import run_xmllint
+from dve.core_engine.backends.utilities import dump_errors
 from dve.core_engine.type_hints import URI, EntityName
 from dve.parser.file_handling import get_content_length, get_parent
 from dve.parser.file_handling.service import open_stream
-from dve.pipeline.utils import dump_errors
 
 SparkXMLMode = Literal["PERMISSIVE", "FAILFAST", "DROPMALFORMED"]
 """The mode to use when parsing XML files with Spark."""
@@ -74,10 +73,11 @@ def __init__(
         trim_cells=True,
         xsd_location: Optional[URI] = None,
         xsd_error_code: Optional[str] = None,
-        xsd_error_message: Optional[str] = None
+        xsd_error_message: Optional[str] = None,
+        rules_location: Optional[URI] = None,
         **_,
     ) -> None:
-        
+
         super().__init__(
             record_tag=record_tag,
             root_tag=root_tag,
@@ -86,7 +86,8 @@ def __init__(
             sanitise_multiline=sanitise_multiline,
             xsd_location=xsd_location,
             xsd_error_code=xsd_error_code,
-            xsd_error_message=xsd_error_message
+            xsd_error_message=xsd_error_message,
+            rules_location=rules_location,
         )
 
         self.spark_session = spark_session or SparkSession.builder.getOrCreate()
@@ -117,16 +118,14 @@ def read_to_dataframe(
         """
         if get_content_length(resource) == 0:
             raise EmptyFileError(f"File at {resource} is empty.")
-        
+
         if self.xsd_location:
             msg = self._run_xmllint(file_uri=resource)
             if msg:
                 working_folder = get_parent(resource)
                 dump_errors(
-                    working_folder=working_folder,
-                    step_name="file_transformation",
-                    messages=[msg]
-                    )
+                    working_folder=working_folder, step_name="file_transformation", messages=[msg]
+                )
 
         spark_schema: StructType = get_type_from_annotation(schema)
         kwargs = {
@@ -165,7 +164,7 @@ def read_to_dataframe(
                     kwargs["rowTag"] = f"{namespace}:{self.record_tag}"
                     df = (
                         self.spark_session.read.format("xml")
-                        .options(**kwargs)
+                        .options(**kwargs)  # type: ignore
                         .load(resource, schema=read_schema)
                     )
             if self.root_tag and df.columns:

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

@@ -36,6 +36,7 @@ def __init__(
         trim_cells: bool = True,
         null_values: Collection[str] = frozenset({"NULL", "null", ""}),
         encoding: str = "utf-8-sig",
+        **_,
     ):
         """Init function for the base CSV reader.
 

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

@@ -103,7 +103,7 @@ def clear(self) -> None:
     def __iter__(self) -> Iterator["XMLElement"]: ...
 
 
-class BasicXMLFileReader(BaseFileReader):
+class BasicXMLFileReader(BaseFileReader):  # pylint: disable=R0902
     """A reader for XML files built atop LXML."""
 
     def __init__(
@@ -119,6 +119,7 @@ def __init__(
         xsd_location: Optional[URI] = None,
         xsd_error_code: Optional[str] = None,
         xsd_error_message: Optional[str] = None,
+        rules_location: Optional[URI] = None,
         **_,
     ):
         """Init function for the base XML reader.
@@ -153,8 +154,11 @@ def __init__(
         """Encoding of the XML file."""
         self.n_records_to_read = n_records_to_read
         """The maximum number of records to read from a document."""
-        self.xsd_location = xsd_location
-        """The relative URI of the xsd file if wishing to perform xsd validation"""
+        if rules_location is not None and xsd_location is not None:
+            self.xsd_location = rules_location + xsd_location
+        else:
+            self.xsd_location = xsd_location  # type: ignore
+            """The URI of the xsd file if wishing to perform xsd validation."""
         self.xsd_error_code = xsd_error_code
         """The error code to be reported if xsd validation fails (if xsd)"""
         self.xsd_error_message = xsd_error_message
@@ -269,12 +273,22 @@ def _parse_xml(
 
         for element in elements:
             yield self._parse_element(element, template_row)
-    
-    def _run_xmllint(self, file_uri: URI) -> FeedbackMessage:
-        return run_xmllint(file_uri=file_uri,
-                        schema_uri=self.xsd_location,
-                        error_code=self.xsd_error_code,
-                        error_message=self.xsd_error_message)
+
+    def _run_xmllint(self, file_uri: URI) -> FeedbackMessage | None:
+        """Run xmllint package to validate against a given xsd. Requires xmlint to be installed
+        onto the system to run succesfully."""
+        if self.xsd_location is None:
+            raise AttributeError("Trying to run XML lint with no `xsd_location` provided.")
+        if self.xsd_error_code is None:
+            raise AttributeError("Trying to run XML with no `xsd_error_code` provided.")
+        if self.xsd_error_message is None:
+            raise AttributeError("Trying to run XML with no `xsd_error_message` provided.")
+        return run_xmllint(
+            file_uri=file_uri,
+            schema_uri=self.xsd_location,
+            error_code=self.xsd_error_code,
+            error_message=self.xsd_error_message,
+        )
 
     def read_to_py_iterator(
         self,

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

@@ -1,11 +1,13 @@
-"""Implement XML linting for files."""
+"""Implement XML linting for files. Please note that xml linting requires xmllint to be installed
+onto your system."""
 
 import shutil
 import tempfile
+from collections.abc import Sequence
 from contextlib import ExitStack
 from pathlib import Path
 from subprocess import PIPE, STDOUT, Popen
-from typing import Sequence, Union
+from typing import Union
 from uuid import uuid4
 
 from dve.core_engine.message import FeedbackMessage
@@ -87,7 +89,7 @@ def run_xmllint(
 
     """
     if not shutil.which("xmllint"):
-        raise OSError("Unable to find `xmllint` binary")
+        raise OSError("Unable to find `xmllint` binary. Please install to use this functionality.")
 
     if not get_resource_exists(file_uri):
         raise IOError(f"No resource accessible at file URI {file_uri!r}")
@@ -128,8 +130,7 @@ def run_xmllint(
                 # Close the input stream and await the response code.
                 # Output will be written to the message file.
                 process.stdin.close()
-                # TODO: Identify an appropriate timeout.
-                return_code = process.wait()
+                return_code = process.wait(10)
 
         if return_code == 0:
             return None

src/dve/core_engine/backends/utilities.py

@@ -1,18 +1,21 @@
 """Necessary, otherwise uncategorised backend functionality."""
 
+import json
 import sys
 from dataclasses import is_dataclass
 from datetime import date, datetime, time
 from decimal import Decimal
 from typing import GenericAlias  # type: ignore
-from typing import Any, ClassVar, Union
+from typing import Any, ClassVar, Optional, Union
 
 import polars as pl  # type: ignore
 from polars.datatypes.classes import DataTypeClass as PolarsType
 from pydantic import BaseModel, create_model
 
+import dve.parser.file_handling as fh
 from dve.core_engine.backends.base.utilities import _get_non_heterogenous_type
-from dve.core_engine.type_hints import Messages
+from dve.core_engine.type_hints import URI, Messages
+from dve.reporting.error_report import conditional_cast
 
 # We need to rely on a Python typing implementation detail in Python <= 3.7.
 if sys.version_info[:2] <= (3, 7):
@@ -176,3 +179,38 @@ def get_polars_type_from_annotation(type_annotation: Any) -> PolarsType:
         if polars_type:
             return polars_type
     raise ValueError(f"No equivalent DuckDB type for {type_annotation!r}")
+
+
+def dump_errors(
+    working_folder: URI,
+    step_name: str,
+    messages: Messages,
+    key_fields: Optional[dict[str, list[str]]] = None,
+):
+    """Write out to disk captured feedback error messages."""
+    if not working_folder:
+        raise AttributeError("processed files path not passed")
+
+    if not key_fields:
+        key_fields = {}
+
+    errors = fh.joinuri(working_folder, "errors", f"{step_name}_errors.json")
+    processed = []
+
+    for message in messages:
+        primary_keys: list[str] = key_fields.get(message.entity if message.entity else "", [])
+        error = message.to_dict(
+            key_field=primary_keys,
+            value_separator=" -- ",
+            max_number_of_values=10,
+            record_converter=None,
+        )
+        error["Key"] = conditional_cast(error["Key"], primary_keys, value_separator=" -- ")
+        processed.append(error)
+
+    with fh.open_stream(errors, "a+") as f:
+        json.dump(
+            processed,
+            f,
+            default=str,
+        )