Merge branch 'main' of https://github.com/NHSDigital/data-validation-engine into build/pyspark_35

Author: georgeRobertson2

CHANGELOG.md

@@ -1,8 +1,27 @@
+## v0.7.6 (2026-04-30)
+
+### Fix
+
+- ddb xml reader connection args consistent with other ddb readers
+- remove default connection from ddb json reader
+- remove default connection in csv reader causing hanging in multithreaded setups
+- configured refdata loader to be instantiated when required without class vars #99
+
+### Refactor
+
+- add in backend kwargs for readers to allow reader args not determinable at config write time to be passed
+
+## v0.7.5 (2026-04-29)
+
+### Fix
+
+- configured refdata loader to be instantiated when required without need for class vars (#98)
+
 ## v0.7.4 (2026-04-17)
 
 ### Fix
 
-- fix failures not being properly audited in foundry pipeline
+- fix failures not being properly audited in foundry pipeline (#94)
 
 ## v0.7.3 (2026-04-16)
 

poetry.lock

@@ -24,7 +24,7 @@ Issues = "https://github.com/NHSDigital/data-validation-engine/issues"
 Changelog = "https://github.com/NHSDigital/data-validation-engine/blob/main/CHANGELOG.md"
 
 [tool.poetry]
-version = "0.7.4"
+version = "0.7.6"
 packages = [
     { include = "dve", from = "src" },
 ]

pyproject.toml

@@ -3,7 +3,7 @@
 import logging
 import warnings
 from abc import ABC, abstractmethod
-from collections.abc import Mapping, MutableMapping
+from collections.abc import MutableMapping
 from typing import Any, ClassVar, Generic, Optional
 
 from pyspark.sql import DataFrame, SparkSession
@@ -41,14 +41,12 @@ def __init__(  # pylint: disable=unused-argument
         self,
         contract: BaseDataContract[EntityType],
         steps: BaseStepImplementations[EntityType],
-        reference_data_loader_type: Optional[type[BaseRefDataLoader[EntityType]]],
         logger: Optional[logging.Logger] = None,
         **kwargs: Any,
     ) -> None:
         for component_name, component in (
             ("Contract", contract),
             ("Step implementation", steps),
-            ("Reference data loader", reference_data_loader_type),
         ):
             component_entity_type = getattr(component, "__entity_type__", None)
             if component_entity_type != self.__entity_type__:
@@ -61,42 +59,16 @@ def __init__(  # pylint: disable=unused-argument
         """The data contract implementation used by the backend."""
         self.step_implementations = steps
         """The step implementations used by the backend."""
-        self.reference_data_loader_type = reference_data_loader_type
-        """
-        The loader type to use for the reference data. If `None`, do not
-        load any reference data and error if it is provided.
-
-        """
         self.logger = logger or get_logger(type(self).__name__)
         """The `logging.Logger instance for the backend."""
 
     def load_reference_data(
         self,
         reference_entity_config: dict[EntityName, ReferenceConfigUnion],
         submission_info: Optional[SubmissionInfo],
-    ) -> Mapping[EntityName, EntityType]:
-        """Load the reference data as specified in the reference entity config."""
-        sub_info_entity: Optional[EntityType] = None
-        if submission_info:
-            sub_info_entity = self.convert_submission_info(submission_info)
-
-        if self.reference_data_loader_type is None:
-            if reference_entity_config:
-                raise ValueError(
-                    "Reference data has been specified but no reference data loader is "
-                    + "configured for this backend"
-                )
-
-            reference_data_dict = {}
-            if sub_info_entity is not None:
-                reference_data_dict["dve_submission_info"] = sub_info_entity
-            return reference_data_dict
-
-        reference_data_loader = self.reference_data_loader_type(reference_entity_config)
-        if sub_info_entity is not None:
-            reference_data_loader.entity_cache["dve_submission_info"] = sub_info_entity
-
-        return reference_data_loader
+    ) -> BaseRefDataLoader[EntityType]:
+        """Supply configured reference data loader for use with business rules"""
+        raise NotImplementedError()
 
     @abstractmethod
     def convert_submission_info(self, submission_info: SubmissionInfo) -> EntityType:

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

@@ -11,6 +11,7 @@
 from dve.core_engine.backends.base.core import get_entity_type
 from dve.core_engine.backends.exceptions import (
     MissingRefDataEntity,
+    NoRefDataConfigSupplied,
     RefdataLacksFileExtensionSupport,
 )
 from dve.core_engine.backends.types import EntityType
@@ -147,11 +148,11 @@ class variable for the subclass.
     # pylint: disable=unused-argument
     def __init__(
         self,
-        reference_entity_config: dict[EntityName, ReferenceConfig],
-        dataset_config_uri: Optional[URI] = None,
+        reference_data_config: dict[EntityName, ReferenceConfig],
+        dataset_config_uri: URI,
         **kwargs,
     ) -> None:
-        self.reference_entity_config = reference_entity_config
+        self.reference_entity_config = reference_data_config
         self.dataset_config_uri = dataset_config_uri
         """
         Configuration options for the reference data. This is likely to vary
@@ -207,6 +208,8 @@ def __getitem__(self, key: EntityName) -> EntityType:
             try:
                 config = self.reference_entity_config[key]
                 return self.load_entity(entity_name=key, config=config)
+            except TypeError as err:
+                raise NoRefDataConfigSupplied() from err
             except Exception as err:
                 raise MissingRefDataEntity(entity_name=key) from err
 

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

@@ -119,6 +119,20 @@ def get_message_preamble(self) -> str:
         return f"Missing reference data entity {self.entity_name!r}"
 
 
+class NoRefDataConfigSupplied(BackendError):
+    """An error raised when trying to load a refdata entity when no refdata
+    config has been supplied.
+
+    """
+
+    def __init__(self, *args: object) -> None:
+        super().__init__(*args)
+
+    def get_message_preamble(self) -> EntityName:
+        """Message for logging purposes"""
+        return "Refdata loader not supplied with refdata config - unable to load refdata entities"
+
+
 class ConstraintError(ValueError, BackendErrorMixin):
     """Raised when a given constraint is violated."""
 

src/dve/core_engine/backends/exceptions.py

@@ -411,7 +411,7 @@ def get_duckdb_cast_statement_from_annotation(
             stmt = rf"CASE WHEN REGEXP_MATCHES(TRIM({quoted_name}), '{date_regex}') THEN TRY_CAST(TRIM({quoted_name}) as DATE) ELSE NULL END"  # pylint: disable=C0301
             return stmt
         if issubclass(type_, time):
-            stmt = rf"CASE WHEN REGEXP_MATCHES(TRIM({quoted_name}), '{time_regex}') THEN TRY_CAST(TRIM({quoted_name}) as TIME) ELSE NULL END" # pylint: disable=C0301
+            stmt = rf"CASE WHEN REGEXP_MATCHES(TRIM({quoted_name}), '{time_regex}') THEN TRY_CAST(TRIM({quoted_name}) as TIME) ELSE NULL END"  # pylint: disable=C0301
             return stmt
         duck_type = get_duckdb_type_from_annotation(type_)
         if duck_type:

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

@@ -6,13 +6,7 @@
 
 import duckdb as ddb
 import polars as pl
-from duckdb import (
-    DuckDBPyConnection,
-    DuckDBPyRelation,
-    StarExpression,
-    default_connection,
-    read_csv,
-)
+from duckdb import DuckDBPyConnection, DuckDBPyRelation, StarExpression, read_csv
 from pydantic import BaseModel
 
 from dve.core_engine.backends.base.reader import BaseFileReader, read_function
@@ -61,7 +55,7 @@ def __init__(
         self.header = header
         self.delim = delim
         self.quotechar = quotechar
-        self._connection = connection if connection else default_connection
+        self._connection = connection if connection else ddb.connect(":memory:")
         self.field_check = field_check
         self.field_check_error_code = field_check_error_code
         self.field_check_error_message = field_check_error_message
@@ -181,7 +175,7 @@ def read_to_relation(  # pylint: disable=unused-argument
             ] + [pl.col(RECORD_INDEX_COLUMN_NAME)]
             df = df.select(pl_exprs)
 
-        return ddb.sql("SELECT * FROM df")
+        return self._connection.sql("SELECT * FROM df")
 
 
 class DuckDBCSVRepeatingHeaderReader(PolarsToDuckDBCSVReader):

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

@@ -4,7 +4,8 @@
 from collections.abc import Iterator
 from typing import Any, Optional
 
-from duckdb import DuckDBPyRelation, read_json
+import duckdb
+from duckdb import DuckDBPyConnection, DuckDBPyRelation
 from pydantic import BaseModel
 
 from dve.core_engine.backends.base.reader import BaseFileReader, read_function
@@ -26,9 +27,11 @@ def __init__(
         self,
         *,
         json_format: Optional[str] = "array",
+        connection: Optional[DuckDBPyConnection] = None,
         **_,
     ):
         self._json_format = json_format
+        self._connection = duckdb.connect(":memory:") if not connection else connection
 
         super().__init__()
 
@@ -50,5 +53,7 @@ def read_to_relation(  # pylint: disable=unused-argument
         }
 
         return self.add_record_index(
-            read_json(resource, columns=ddb_schema, format=self._json_format)  # type: ignore
+            self._connection.read_json(
+                resource, columns=ddb_schema, format=self._json_format  # type: ignore
+            )
         )

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

@@ -3,8 +3,9 @@
 
 from typing import Optional
 
+import duckdb
 import polars as pl
-from duckdb import DuckDBPyConnection, DuckDBPyRelation, default_connection
+from duckdb import DuckDBPyConnection, DuckDBPyRelation
 from pydantic import BaseModel
 
 from dve.core_engine.backends.base.reader import read_function
@@ -24,8 +25,8 @@
 class DuckDBXMLStreamReader(XMLStreamReader):
     """A reader for XML files"""
 
-    def __init__(self, *, ddb_connection: Optional[DuckDBPyConnection] = None, **kwargs):
-        self.ddb_connection = ddb_connection if ddb_connection else default_connection
+    def __init__(self, *, connection: Optional[DuckDBPyConnection] = None, **kwargs):
+        self._connection = connection if connection else duckdb.connect(":memory:")
         super().__init__(**kwargs)
 
     @read_function(DuckDBPyRelation)
@@ -49,4 +50,4 @@ def read_to_relation(self, resource: URI, entity_name: str, schema: type[BaseMod
                 data=self.read_to_py_iterator(resource, entity_name, schema), schema=polars_schema
             )
         )
-        return self.ddb_connection.sql("select * from _lazy_frame")
+        return self._connection.sql("select * from _lazy_frame")