Merge branch 'release_v010' of https://github.com/NHSDigital/data-validation-engine into feature/gr-ndsp-619-add_group_level_rejections

Author: georgeRobertson

docs/advanced_guidance/json_schemas/entity_relationships.schema.json

@@ -20,10 +20,10 @@
                 "mandatory": {
                     "type": "boolean"
                 },
-                "orphaned_records_error_code": {
+                "missing_parent_id_error_code": {
                     "type": "string"
                 },
-                "orphaned_records_error_message": {
+                "missing_parent_id_error_message": {
                     "type": "string"
                 }
             },

src/dve/common/error_utils.py

@@ -5,7 +5,7 @@
 import logging
 from collections.abc import Iterable
 from itertools import chain
-from multiprocessing import Queue
+from queue import Queue
 from threading import Thread
 from typing import Optional, Union
 

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

@@ -46,11 +46,7 @@
     TableUnion,
 )
 from dve.core_engine.backends.types import Entities, EntityType, StageSuccessful
-from dve.core_engine.configuration.v1.hierarchy import (
-    ChildHierarchyNode,
-    EntityHierarchy,
-    HierarchyNode,
-)
+from dve.core_engine.configuration.v1.hierarchy import EntityHierarchy, HierarchyNode
 from dve.core_engine.constants import ORPHANED_RECORD_ENTITY_NAME
 from dve.core_engine.exceptions import CriticalProcessingError
 from dve.core_engine.loggers import get_logger
@@ -399,7 +395,7 @@ def identify_and_remove_orphans(
         """
 
         def process_node(
-            node: HierarchyNode | ChildHierarchyNode,
+            node: HierarchyNode,
             parent_entity_name: Optional[EntityName],
             orph_messages: Messages | None = None,
         ):
@@ -409,7 +405,7 @@ def process_node(
             if orph_messages is None:
                 orph_messages = []
 
-            if isinstance(node, ChildHierarchyNode) and parent_entity_name is not None:
+            if parent_entity_name is not None:
                 self.logger.info(f"Identifying orphans in {current_entity_name}")
 
                 join_expr = " AND ".join(
@@ -428,7 +424,9 @@ def process_node(
                 )
 
                 if no_orphs > 0:
-                    self.logger.info(f"Removing orphan records from {current_entity_name}")
+                    self.logger.info(
+                        f"Removing records with missing parent from {current_entity_name}"
+                    )
                     location = list(node.join_fields.values())[0]
                     with BackgroundMessageWriter(
                         working_directory=working_directory,
@@ -442,28 +440,29 @@ def process_node(
                                 entity_name=current_entity_name,
                                 reporting=ReportingConfig(
                                     emit="record_failure",
-                                    code=node.orphaned_records_error_code,
-                                    message=node.orphaned_records_error_message,
+                                    code=node.missing_parent_id_error_code,
+                                    message=node.missing_parent_id_error_message,
                                     location=location,
                                 ),
                             ),
                         )
-                        for record in _orph_records:
-                            msg_writer.write_queue.put(
-                                [
-                                    FeedbackMessage(
-                                        entity=current_entity_name,
-                                        record=record,  # type: ignore
-                                        error_location=location,
-                                        error_message=node.orphaned_records_error_message,
-                                        failure_type="record",
-                                        error_type="record",
-                                        error_code=node.orphaned_records_error_code,
-                                        reporting_field=location,
-                                        category="Parent Missing",
-                                    )
-                                ]
-                            )
+                        # moved to batch the write - risky if large number of
+                        msg_writer.write_queue.put(
+                            [
+                                FeedbackMessage(
+                                    entity=current_entity_name,
+                                    record=record,  # type: ignore
+                                    error_location=location,
+                                    error_message=node.missing_parent_id_error_message,
+                                    failure_type="record",
+                                    error_type="record",
+                                    error_code=node.missing_parent_id_error_code,
+                                    reporting_field=location,
+                                    category="Parent Missing",
+                                )
+                                for record in _orph_records
+                            ]
+                        )
 
             if node.children:
                 for child_node in node.children:

src/dve/core_engine/configuration/v1/__init__.py

@@ -3,7 +3,8 @@
 import json
 from typing import Any, Optional, Type, Union
 
-from pydantic import BaseModel, Field, PrivateAttr, validate_call
+from pydantic import BaseModel, Field, PrivateAttr, field_validator, model_validator, validate_call
+from pydantic_core.core_schema import FieldValidationInfo
 from typing_extensions import Literal
 
 from dve.core_engine.backends.base.reference_data import ReferenceConfig, ReferenceConfigUnion
@@ -93,23 +94,48 @@ class _TypeAliasDefinition(_BaseTypeDefintion):
 class _LinkageConfig(BaseModel):
     """Specify how to link entities back to parents if required"""
 
-    parent_entity: EntityName
+    parent_entity: Optional[EntityName] = None
     """The name of the parent entity"""
-    join_fields: JoinFields
+    join_fields: JoinFields = Field(default_factory=dict)
     """The fields that can be used to link back to the parent entity"""
-    mandatory: Optional[bool] = False
+    is_root_entity: bool = False
+    """Whether the entity is the highest level parent in a tree"""
+    mandatory: bool = False
     """If the entity is a child, is it a mandatory field of the parent"""
     no_valid_records_error_code: Optional[ErrorCode] = "NoValidRecords"
     """The error code to emit if the entity has no valid records and is mandatory in the parent entity"""  # pylint: disable=C0301
     no_valid_records_error_message: Optional[ErrorMessage] = (
         "parent record removed as no valid child records"
     )
     """The error message to emit if the entity has no valid records and is mandatory in the parent entity"""  # pylint: disable=C0301
-    orphaned_records_error_code: Optional[ErrorCode] = "OrphanedRecords"
+    missing_parent_id_error_code: Optional[ErrorCode] = "MissingParentRecord"
     """The error code to emit if the entity contains records that are orphaned by parent record rejections"""  # pylint: disable=C0301
-    orphaned_records_error_message: Optional[ErrorMessage] = "Orphaned records removed"
+    missing_parent_id_error_message: Optional[ErrorMessage] = (
+        "Records removed due to no valid parent record"
+    )
     """The error code to emit if the entity contains records that are orphaned by parent record rejections"""  # pylint: disable=C0301
 
+    @model_validator(mode="after")
+    def _check_root_no_parent_or_join_keys(self):
+        if self.is_root_entity and (self.parent_entity or self.join_fields):
+            raise ValueError(
+                "If entity is root, neither parent_entity nor join keys should be specified"
+            )
+        return self
+
+    @model_validator(mode="after")
+    def _check_root_mandatory(self):
+        if self.is_root_entity and not self.mandatory:
+            raise ValueError("If entity is root, it must be labelled mandatory")
+        return self
+
+    @model_validator(mode="after")
+    def _check_parent_entity_with_join_keys(self):
+        if self.parent_entity or self.join_fields:
+            if not (self.parent_entity and self.join_fields):
+                raise ValueError("Both parent_entity and join_fields must be supplied if one is")
+        return self
+
 
 class _SchemaConfig(BaseModel):
     """Configuration for a component schema within a dataset."""

src/dve/core_engine/configuration/v1/hierarchy.py

@@ -16,7 +16,15 @@ class HierarchyNode(BaseModel):
     """Stores entity hierarchy information"""
 
     entity_name: str
-    children: Optional[list["ChildHierarchyNode"]] = Field(default_factory=list)
+    children: list["HierarchyNode"] = Field(default_factory=list)
+    mandatory: bool = False
+    join_fields: dict[str, str] = Field(default_factory=dict)
+    no_valid_records_error_code: ErrorCode = "NoValidRecords"
+    no_valid_records_error_message: ErrorMessage = "parent record removed as no valid child records"
+    missing_parent_id_error_code: Optional[ErrorCode] = "MissingParentRecord"
+    missing_parent_id_error_message: Optional[ErrorMessage] = (
+        "Records removed due to no valid parent record"
+    )
 
     def get_descendents(self) -> list[str]:
         """Recursively list all descendents of the node"""
@@ -58,19 +66,6 @@ def as_dict(self) -> dict[str, dict[str, Any]]:
         return {self.entity_name: ret_dict}
 
 
-class ChildHierarchyNode(HierarchyNode):
-    """Stores child entity hierarchy information"""
-
-    join_fields: dict[str, str]
-    mandatory: Optional[bool] = False
-    no_valid_records_error_code: Optional[ErrorCode] = "NoValidRecords"
-    no_valid_records_error_message: Optional[ErrorMessage] = (
-        "parent record removed as no valid child records"
-    )
-    orphaned_records_error_code: Optional[ErrorCode] = "OrphanedRecords"
-    orphaned_records_error_message: Optional[ErrorMessage] = "Orphaned records removed"
-
-
 class EntityHierarchy:
     """Determines and stores entity hierarchy information from config"""
 
@@ -82,12 +77,35 @@ def determine_trees(
         all_datasets: Iterable[str], entity_relationships: dict[str, _LinkageConfig]
     ) -> dict[EntityName, HierarchyNode]:
         """Determine the entity hierarchy trees and store as HierarchyNodes"""
+        root_entities: dict[str, _LinkageConfig] = dict(
+            filter(lambda x: x[1].is_root_entity, entity_relationships.items())
+        )
         top_level_parents: dict[EntityName, HierarchyNode] = {
-            entity_name: HierarchyNode(entity_name=entity_name)
-            for entity_name in all_datasets
-            if entity_name not in entity_relationships
+            entity_name: HierarchyNode(
+                entity_name=entity_name,
+                **config.model_dump(
+                    exclude={
+                        "parent_entity",
+                        "missing_parent_id_error_code",
+                        "missing_parent_id_error_message",
+                    }
+                ),
+                missing_parent_id_error_code=None,
+                missing_parent_id_error_message=None,
+            )
+            for entity_name, config in root_entities.items()
         }
 
+        if default_roots := [
+            entity_name for entity_name in all_datasets if entity_name not in entity_relationships
+        ]:
+            for entity_name in default_roots:
+                top_level_parents[entity_name] = HierarchyNode(
+                    entity_name=entity_name,
+                    missing_parent_id_error_code=None,
+                    missing_parent_id_error_message=None,
+                )
+
         for name, linkage_detail in entity_relationships.items():
             for main_entity, parent_node in top_level_parents.items():
                 if (
@@ -96,7 +114,7 @@ def determine_trees(
                 ):
                     parent_node.add_child_node(
                         linkage_detail.parent_entity,
-                        ChildHierarchyNode(
+                        HierarchyNode(
                             entity_name=name, **linkage_detail.model_dump(exclude={"parent_entity"})
                         ),
                     )

src/dve/core_engine/constants.py

@@ -7,5 +7,5 @@
 """The name of the field that can be used to extract the field value that caused
    a pydantic validation error"""
 
-ORPHANED_RECORD_ENTITY_NAME: str = "orphaned_records_tracker"
-"""Name to keep track of identified orphaned records"""
+ORPHANED_RECORD_ENTITY_NAME: str = "orphaned_record_tracker"
+"""Name of entity to keep track of records where there is a missing parent record"""

src/dve/core_engine/models.py

@@ -82,7 +82,9 @@ def _ensure_just_file_stem(
     @property
     def file_name_with_ext(self):
         """Return file name with extension."""
-        return f"{self.file_name}.{self.file_extension}"
+        if self.file_extension:
+            return f"{self.file_name}.{self.file_extension}"
+        return self.file_name
 
     @classmethod
     def from_metadata_file(cls, submission_id: str, metadata_uri: Location):

src/dve/pipeline/pipeline.py

@@ -216,10 +216,10 @@ def write_file_to_parquet(
 
         for model_name, model in models.items():
             self._logger.info(f"Transforming {model_name} to stringified parquet")
-            reader: BaseFileReader = load_reader(
-                dataset, model_name, ext, self.backend_reader_kwargs
-            )
             try:
+                reader: BaseFileReader = load_reader(
+                    dataset, model_name, ext, self.backend_reader_kwargs
+                )
                 if not entity_type:
                     reader.write_parquet(
                         reader.read_to_py_iterator(
@@ -242,6 +242,7 @@ def write_file_to_parquet(
                         f"{out}{model_name}",
                     )
             except MessageBearingError as exc:
+                self._logger.error(f"Unable to process {model_name}", exc_info=exc)
                 errors.extend(exc.messages)
 
         return list(dict.fromkeys(errors))  # remove any duplicate errors

src/dve/pipeline/utils.py

@@ -11,9 +11,11 @@
 import dve.core_engine.backends.implementations.duckdb  # pylint: disable=unused-import
 import dve.core_engine.backends.implementations.spark  # pylint: disable=unused-import
 import dve.parser.file_handling as fh
+from dve.core_engine.backends.exceptions import MessageBearingError
 from dve.core_engine.backends.readers import _READER_REGISTRY
 from dve.core_engine.configuration.v1 import SchemaName, V1EngineConfig, _ModelConfig
 from dve.core_engine.loggers import get_logger
+from dve.core_engine.message import FeedbackMessage
 from dve.core_engine.type_hints import URI, SubmissionResult
 from dve.metadata_parser.model_generator import JSONtoPyd
 
@@ -52,7 +54,31 @@ def load_reader(
     backend_reader_kwargs: Optional[dict[str, Any]] = None,
 ):
     """Loads the readers for the diven feed, model name and file extension"""
-    reader_config = dataset[model_name].reader_config[f".{file_extension.lower()}"]
+    try:
+        reader_config = dataset[model_name].reader_config[f".{file_extension.lower()}"]
+    except KeyError as exc:
+        if file_extension:
+            err_msg = (
+                f"The supplied file extension `{file_extension}`"
+                +f" is not a supported file format for {model_name}."
+            )
+        else:
+            err_msg = "No supplied file extension. Unable to parse file without a file extension."
+
+        raise MessageBearingError(
+            f"The file extension provided ({file_extension}) is not supported for this collection.",
+            messages=[
+                FeedbackMessage(
+                    entity=model_name,
+                    record=None,
+                    failure_type="submission",
+                    error_location="Whole File",
+                    error_code="InvalidFileExtension",
+                    error_message=err_msg,
+                )
+            ],
+        ) from exc
+
     reader = _READER_REGISTRY[reader_config.reader](
         **reader_config.kwargs_, **backend_reader_kwargs if backend_reader_kwargs else {}
     )

tests/features/planets.feature

@@ -43,7 +43,9 @@ Feature: Pipeline tests using the planets dataset
         And I add initial audit entries for the submission
         Then the latest audit record for the submission is marked with processing status file_transformation
         When I run the file transformation phase
-        Then the latest audit record for the submission is marked with processing status failed
+        Then the latest audit record for the submission is marked with processing status error_report
+        When I run the error report phase
+        Then An error report is produced
 
     Scenario: Handle a file with duplicated extension provided (spark)
         Given I submit the planets file planets.csv.csv for processing