feat: added support for nested fields when configuring custom error details for data contract. Includes accessing nested error values in error messages.

Author: stevenhsd

src/dve/core_engine/message.py

@@ -1,9 +1,12 @@
 """Functionality to represent messages."""
 
+import copy
 import datetime as dt
+from functools import reduce
 import json
 from decimal import Decimal
-from typing import Any, Callable, ClassVar, Dict, List, Optional, Set, Type, Union
+import operator
+from typing import Any, Callable, ClassVar, Dict, List, Optional, Set, Tuple, Type, Union
 
 from pydantic import ValidationError, validator, BaseModel
 from pydantic.dataclasses import dataclass
@@ -24,16 +27,28 @@
 class DataContractErrorDetail(BaseModel):
     error_code: str
     error_message: Optional[str] = None
-    def template_message(self, variables: Dict[str, Any]) -> str:
+    def template_message(self,
+                         variables: Dict[str, Any],
+                         error_location: Optional[Tuple[Union[str, int], ...]] = None) -> str:
+        if error_location:
+            variables = self.extract_error_value(variables, error_location)
         return template_object(self.error_message, variables)
-
+    @staticmethod
+    def extract_error_value(records, error_location):
+        _records = copy.copy(records)
+        try:
+            _records["__error_value"] = reduce(operator.getitem, error_location, _records)
+        except KeyError:
+            pass
+        return _records
+        
 DEFAULT_ERROR_DETAIL: Dict[ErrorCategory, DataContractErrorDetail] = {
     "Blank": DataContractErrorDetail(error_code="FieldBlank",
                                      error_message="cannot be blank"),
     "Bad value": DataContractErrorDetail(error_code="BadValue",
                                          error_message="is invalid"),
     "Wrong Format": DataContractErrorDetail(error_code="WrongFormat",
-                                         error_message="has worng format"),}
+                                         error_message="has wrong format"),}
 
 
 INTEGRITY_ERROR_CODES: Set[str] = {"blockingsubmission"}
@@ -190,12 +205,15 @@ def from_pydantic_error(
                 failure_type = "submission"
             else:
                 failure_type = "record"
+            
+            error_field = ".".join([idx for idx in error_dict["loc"]
+                                    if not isinstance(idx, int)])
 
             is_informational = False
             if error_code.endswith("warning"):
                 is_informational = True
             error_detail: DataContractErrorDetail = error_details.get(
-                error_dict["loc"][-1],
+                error_field,
                 DEFAULT_ERROR_DETAIL
                 ).get(category)
 
@@ -207,7 +225,7 @@ def from_pydantic_error(
                     is_informational=is_informational,
                     error_type=error_type,
                     error_location=error_dict["loc"],  # type: ignore
-                    error_message=error_detail.template_message(record),
+                    error_message=error_detail.template_message(record, error_dict["loc"]),
                     reporting_field=error_dict["loc"][-1],  # type: ignore
                     category=category,  # type: ignore
                     error_code=error_detail.error_code,  # type: ignore

src/dve/core_engine/templating.py

@@ -83,7 +83,7 @@ def template_object(
 
     if isinstance(object_, str):
         if method == "jinja":
-            return ENVIRONMENT.from_string(object_).render(**variables)  # type: ignore
+            return ENVIRONMENT.from_string(object_).render(variables)  # type: ignore
         return object_.format(**variables)  # type: ignore
 
     parameterise = partial(template_object, variables=variables, method=method)

tests/test_core_engine/test_backends/fixtures.py

@@ -461,39 +461,129 @@ def nested_typecast_parquet(temp_dir) -> Iterator[Tuple[URI, List[Dict[str, Any]
     _df.coalesce(1).write.format("parquet").save(output_location)
     yield output_location, data
 
+@pytest.fixture(scope="function")
+def nested_all_string_parquet_w_errors(temp_dir,
+                                       nested_parquet_custom_dc_err_details) -> Iterator[Tuple[URI, str, List[Dict[str, Any]]]]:
+    contract_meta = json.dumps(
+        {
+            "contract": {
+                "error_details": f"{nested_parquet_custom_dc_err_details.as_posix()}",
+                "schemas": {
+                    "SubField": {
+                        "fields": {
+                            "id": "int",
+                            "substrfield": "str",
+                            "subarrayfield": {"type": "date", "is_array": True},
+                        },
+                        "mandatory_fields": ["id"],
+                    }
+                },
+                "datasets": {
+                    "nested_model": {
+                        "fields": {
+                            "id": "int",
+                            "strfield": "str",
+                            "datetimefield": "datetime",
+                            "subfield": {"model": "SubField", "is_array": True},
+                        },
+                        "reader_config": {
+                            ".xml": {
+                                "reader": "DuckDBXMLStreamReader",
+                                "parameters": {"root_tag": "root", "record_tag": "NestedModel"},
+                            }
+                        },
+                        "key_field": "id",
+                    }
+                },
+            }
+        }
+    )
+
+    _spark: SparkSession = SparkSession.builder.getOrCreate()
+    data: List[Dict[str, Any]] = [
+        dict(
+            id=1,
+            strfield="hi",
+            datetimefield=str(datetime(2020, 9, 20, 12, 34, 56)),
+            subfield=[
+                dict(
+                    id=1,
+                    substrfield="bye",
+                    subarrayfield=[str(date(2020, 9, 20)), str(date(2020, 9, 21))],
+                )
+            ],
+        ),
+        dict(
+            id="WRONG",
+            strfield="hello",
+            datetimefield=str(datetime(2020, 9, 21, 12, 34, 56)),
+            subfield=[
+                dict(
+                    id=2,
+                    substrfield="bye",
+                    subarrayfield=[str(date(2020, 9, 20)), str(date(2020, 9, 21))],
+                ),
+                dict(
+                    id="WRONG",
+                    substrfield="aurevoir",
+                    subarrayfield=[str(date(2020, 9, 22)), str(date(2020, 9, 23))],
+                ),
+            ],
+        ),
+    ]
+
+    output_location: URI = str(Path(temp_dir).joinpath("nested_parquet").as_posix()) + "/"
+
+    _df: DataFrame = _spark.createDataFrame(
+        data,
+        schema=StructType(
+            [
+                StructField("id", StringType()),
+                StructField("strfield", StringType()),
+                StructField("datetimefield", StringType()),
+                StructField(
+                    "subfield",
+                    ArrayType(
+                        StructType(
+                            [
+                                StructField("id", StringType()),
+                                StructField("substrfield", StringType()),
+                                StructField("subarrayfield", ArrayType(StringType())),
+                            ]
+                        )
+                    ),
+                ),
+            ]
+        ),
+    )
+    _df.coalesce(1).write.format("parquet").save(output_location)
+    yield output_location, contract_meta, data
+
+
 @pytest.fixture()
 def nested_parquet_custom_dc_err_details(temp_dir):
+    file_path = Path(temp_dir).joinpath("nested_parquet_data_contract_codes.json")
     err_details = {
         "id": {
-            "Blank": {"error_code": "TESTID",
+            "Blank": {"error_code": "TESTIDBLANK",
                       "error_message": "id cannot be null"},
-            "Bad Value": {"error_code": "TESTID",
+            "Bad value": {"error_code": "TESTIDBAD",
                           "error_message": "id is invalid: id - {{id}}"}
                 },
         "datetimefield": {
-            "Bad Value": {"error_code": "TESTDTFIELD",
+            "Bad value": {"error_code": "TESTDTFIELDBAD",
                           "error_message": "datetimefield is invalid: id - {{id}}, datetimefield - {{datetimefield}}"}
-        }
+        },
+        "subfield.id": {
+            "Blank": {"error_code": "SUBFIELDTESTIDBLANK",
+                      "error_message": "subfield id cannot be null"},
+            "Bad value": {"error_code": "SUBFIELDTESTIDBAD",
+                          "error_message": "subfield id is invalid: subfield.id - {{__error_value}}"}
+                },
             }
+    with open(file_path, mode="w") as fle:
+        json.dump(err_details, fle)
+    
+    yield file_path
 
-
-
-StructType(
-        [
-            StructField("id", StringType()),
-            StructField("strfield", StringType()),
-            StructField("datetimefield", StringType()),
-            StructField(
-                "subfield",
-                ArrayType(
-                    StructType(
-                        [
-                            StructField("id", StringType()),
-                            StructField("substrfield", StringType()),
-                            StructField("subarrayfield", ArrayType(StringType())),
-                        ]
-                    )
-                ),
-            ),
-        ]
-    )
+    

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

@@ -19,6 +19,7 @@
 from tests.test_core_engine.test_backends.fixtures import (
     nested_all_string_parquet,
     simple_all_string_parquet,
+    nested_all_string_parquet_w_errors,
     nested_parquet_custom_dc_err_details,
     temp_csv_file,
     temp_duckdb_dir,
@@ -307,5 +308,48 @@ def test_ddb_data_contract_read_nested_parquet(nested_all_string_parquet):
         "subfield": "STRUCT(id BIGINT, substrfield VARCHAR, subarrayfield DATE[])[]",
     }
 
-def test_ddb_data_contract_custom_error_details():
-    pass
+def test_duckdb_data_contract_custom_error_details(nested_all_string_parquet_w_errors,
+                                                  nested_parquet_custom_dc_err_details):
+    parquet_uri, contract_meta, _ = nested_all_string_parquet_w_errors
+    connection = default_connection
+    data_contract = DuckDBDataContract(connection)
+
+    entity = data_contract.read_parquet(path=parquet_uri)
+    assert entity.count("*").fetchone()[0] == 2
+   
+    # check processes entity
+    contract_dict = json.loads(contract_meta).get("contract")
+    entities: Dict[str, DuckDBPyRelation] = {
+        "nested_model": entity,
+    }
+    
+    with open(nested_parquet_custom_dc_err_details) as err_dets:
+        custom_error_details = json.load(err_dets)
+
+    dc_meta = DataContractMetadata(
+        reader_metadata={
+            "nested_model": {
+                ".xml": ReaderConfig(
+                    **contract_dict.get("datasets", {})
+                    .get("nested_model", {})
+                    .get("reader_config", {})
+                    .get(".xml")
+                )
+            }
+        },
+        validators={
+            "nested_model": RowValidator(contract_dict,
+                                         "nested_model",
+                                         error_info=custom_error_details)
+        },
+        reporting_fields={"nested_model": ["id"]},
+    )
+
+    entities, messages, stage_successful = data_contract.apply_data_contract(entities, dc_meta)
+    assert stage_successful
+    assert len(messages) == 2
+    messages = sorted(messages, key= lambda x: x.error_code)
+    assert messages[0].error_code == "SUBFIELDTESTIDBAD"
+    assert messages[0].error_message == "subfield id is invalid: subfield.id - WRONG"
+    assert messages[1].error_code == "TESTIDBAD"
+    assert messages[1].error_message == "id is invalid: id - WRONG"

tests/test_core_engine/test_backends/test_implementations/test_spark/test_data_contract.py

@@ -20,6 +20,7 @@
 from dve.core_engine.validation import RowValidator
 from tests.test_core_engine.test_backends.fixtures import (
     nested_all_string_parquet,
+    nested_all_string_parquet_w_errors,
     simple_all_string_parquet,
     nested_parquet_custom_dc_err_details
 )
@@ -172,12 +173,11 @@ def test_spark_data_contract_read_nested_parquet(nested_all_string_parquet):
         ]
     )
 
-def test_spark_data_contract_custom_error_details(nested_all_string_parquet):
-    # can we read in a stringified parquet and run the data contract on it?
-    # more complex file - nested, arrays of structs
-    parquet_uri, contract_meta, _ = nested_all_string_parquet
+def test_spark_data_contract_custom_error_details(nested_all_string_parquet_w_errors,
+                                                  nested_parquet_custom_dc_err_details):
+    parquet_uri, contract_meta, _ = nested_all_string_parquet_w_errors
     data_contract = SparkDataContract()
-    # check can read
+
     entity = data_contract.read_parquet(path=parquet_uri)
     assert entity.count() == 2
     assert entity.schema == StructType(
@@ -204,6 +204,9 @@ def test_spark_data_contract_custom_error_details(nested_all_string_parquet):
     entities: Dict[str, DataFrame] = {
         "nested_model": entity,
     }
+    
+    with open(nested_parquet_custom_dc_err_details) as err_dets:
+        custom_error_details = json.load(err_dets)
 
     dc_meta = DataContractMetadata(
         reader_metadata={
@@ -217,40 +220,20 @@ def test_spark_data_contract_custom_error_details(nested_all_string_parquet):
             }
         },
         validators={
-            "nested_model": RowValidator(contract_dict, "nested_model"),
+            "nested_model": RowValidator(contract_dict,
+                                         "nested_model",
+                                         error_info=custom_error_details)
         },
         reporting_fields={"nested_model": ["id"]},
     )
 
     entities, messages, stage_successful = data_contract.apply_data_contract(entities, dc_meta)
     assert stage_successful
-    assert len(messages) == 0
-    assert entities["nested_model"].count() == 2
-    # check writes entity to parquet
-    output_path: Path = Path(parquet_uri).parent.joinpath("nested_model_output.parquet")
-    data_contract.write_parquet(
-        entity=entities["nested_model"], target_location=output_path.as_posix()
-    )
-    assert output_path.exists()
-    # check when read back in what is expected
-    check = data_contract.read_parquet(path=output_path.as_posix())
-    assert check.count() == 2
-    assert check.schema == StructType(
-        [
-            StructField("id", LongType()),
-            StructField("strfield", StringType()),
-            StructField("datetimefield", TimestampType()),
-            StructField(
-                "subfield",
-                ArrayType(
-                    StructType(
-                        [
-                            StructField("id", LongType()),
-                            StructField("substrfield", StringType()),
-                            StructField("subarrayfield", ArrayType(DateType())),
-                        ]
-                    )
-                ),
-            ),
-        ]
-    )
+    assert len(messages) == 2
+    messages = sorted(messages, key= lambda x: x.error_code)
+    assert messages[0].error_code == "SUBFIELDTESTIDBAD"
+    assert messages[0].error_message == "subfield id is invalid: subfield.id - WRONG"
+    assert messages[1].error_code == "TESTIDBAD"
+    assert messages[1].error_message == "id is invalid: id - WRONG"
+
+