Merge pull request #16 from NHSDigital/develop_v04

merge v0.4.0

Author: stevenhsd

.github/workflows/ci_testing.yml

@@ -17,7 +17,7 @@ jobs:
       - name: Install extra dependencies for a python install
         run: |
           sudo apt-get update
-          sudo apt -y install --no-install-recommends liblzma-dev libbz2-dev libreadline-dev
+          sudo apt -y install --no-install-recommends liblzma-dev libbz2-dev libreadline-dev libxml2-utils
 
       - name: Install asdf cli
         uses: asdf-vm/actions/setup@v4

CHANGELOG.md

@@ -1,3 +1,18 @@
+## v0.4.0 (2025-12-17)
+
+### Feat
+
+- add persistance of error aggregates to pipeline
+- add Foundry pipeline
+
+### Fix
+
+- issue where templated error messages would not correctly format when passing in parameter values
+
+### Refactor
+
+- include submission status for services passthrough
+
 ## v0.3.0 (2025-11-19)
 
 ### Feat

docs/README.md

@@ -234,10 +234,10 @@ audit_manager = SparkAuditingManager(
 
 # Setting up the Pipeline (in this case the Spark implemented one)
 pipeline = SparkDVEPipeline(
+    processed_files_path="path/where/my/processed_files/should_go/",
     audit_tables=audit_manager,
     job_run_id=1,
     rules_path="path/to/my_dischema",
-    processed_files_path="path/where/my/processed_files/should_go/",
     submitted_files_path="path/to/my/cwt_files/",
     reference_data_loader=SparkParquetRefDataLoader,
     spark=spark

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "nhs_dve"
-version = "0.3.0"
+version = "0.4.0"
 description = "`nhs data validation engine` is a framework used to validate data"
 authors = ["NHS England <england.contactus@nhs.net>"]
 readme = "README.md"
@@ -39,7 +39,7 @@ requests = "2.32.4"  # Mitigates security vuln in < 2.31.0
 schedula = "1.2.19"
 sqlalchemy = "2.0.19"
 typing_extensions = "4.6.2"
-urllib3 = "2.5.0"  # Mitigates security vuln in < 1.26.19
+urllib3 = "2.6.0"  # Mitigates security vuln in < 2.5.0
 xmltodict = "0.13.0"
 
 [tool.poetry.group.dev]

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

@@ -31,6 +31,7 @@
     QueueType,
     SubmissionResult,
 )
+from dve.pipeline.utils import SubmissionStatus
 
 AuditReturnType = TypeVar("AuditReturnType")  # pylint: disable=invalid-name
 
@@ -329,7 +330,7 @@ def mark_business_rules(self, submissions: list[tuple[str, bool]], **kwargs):
             ProcessingStatusRecord(
                 submission_id=submission_id,
                 processing_status="business_rules",
-                submission_result="failed" if failed else None,
+                submission_result="validation_failed" if failed else None,
                 **kwargs,
             )
             for submission_id, failed in submissions
@@ -379,7 +380,10 @@ def mark_failed(self, submissions: list[str], **kwargs):
         """Update submission processing_status to failed."""
         recs = [
             ProcessingStatusRecord(
-                submission_id=submission_id, processing_status="failed", **kwargs
+                submission_id=submission_id,
+                processing_status="failed",
+                submission_result="processing_failed",
+                **kwargs,
             )
             for submission_id in submissions
         ]
@@ -494,6 +498,33 @@ def get_submission_statistics(self, submission_id: str) -> Optional[SubmissionSt
         except StopIteration:
             return None
 
+    def get_submission_status(self, submission_id: str) -> Optional[SubmissionStatus]:
+        """Get the latest submission status for a submission"""
+
+        try:
+            processing_rec: ProcessingStatusRecord = next(  # type: ignore
+                self._processing_status.conv_to_records(
+                    self._processing_status.get_most_recent_records(
+                        order_criteria=[OrderCriteria("time_updated", True)],
+                        pre_filter_criteria=[FilterCriteria("submission_id", submission_id)],
+                    )
+                )
+            )
+        except StopIteration:
+            return None
+        sub_status = SubmissionStatus()
+        sub_stats_rec: Optional[SubmissionStatisticsRecord] = self.get_submission_statistics(
+            submission_id
+        )
+        if processing_rec.submission_result == "processing_failed":
+            sub_status.processing_failed = True
+        if processing_rec.submission_result == "validation_failed":
+            sub_status.validation_failed = True
+        if sub_stats_rec:
+            sub_status.number_of_records = sub_stats_rec.record_count
+
+        return sub_status
+
     def __enter__(self):
         """Use audit table as context manager"""
         if self.pool and self.pool_result.done():

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,6 +8,7 @@
 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
@@ -18,13 +19,21 @@
 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)
 
     @read_function(DuckDBPyRelation)
     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)
+            if msg:
+                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