Error/warn when an illegal data type is discovered (#367)

* Implement get_concrete_data_types

* Detect when a variable data type is unsupported and error/warn

* Make exception publicly available

Author: mmwinther

src/dapla_metadata/datasets/__init__.py

@@ -2,6 +2,7 @@
 
 from datadoc_model.all_optional import model
 
+from ._merge import InconsistentDatasetsError
 from ._merge import InconsistentDatasetsWarning
 from .core import Datadoc
 from .dapla_dataset_path_info import DaplaDatasetPathInfo

src/dapla_metadata/datasets/_merge.py

@@ -189,21 +189,25 @@ def check_variables_consistency(
     return results
 
 
-def check_ready_to_merge(
-    results: list[DatasetConsistencyStatus], *, errors_as_warnings: bool
+def report_metadata_consistency(
+    results: list[DatasetConsistencyStatus],
+    *,
+    errors_as_warnings: bool,
+    message: str = INCONSISTENCIES_MESSAGE,
 ) -> None:
     """Check if the datasets are consistent enough to make a successful merge of metadata.
 
     Args:
         results: List if dict with property name and boolean success flag
         errors_as_warnings: True if failing checks should be raised as warnings, not errors.
+        message: The primary message to be displayed.
 
     Raises:
         InconsistentDatasetsError: If inconsistencies are found and `errors_as_warnings == False`
     """
     if failures := [result for result in results if not result.success]:
         messages_list = "\n - ".join(str(f) for f in failures)
-        msg = f"{INCONSISTENCIES_MESSAGE}\n - {messages_list}"
+        msg = f"{message}\n - {messages_list}"
         if errors_as_warnings:
             warnings.warn(
                 message=msg,

src/dapla_metadata/datasets/core.py

@@ -18,13 +18,18 @@
 from dapla_metadata.dapla import user_info
 from dapla_metadata.datasets._merge import DatasetConsistencyStatus
 from dapla_metadata.datasets._merge import check_dataset_consistency
-from dapla_metadata.datasets._merge import check_ready_to_merge
 from dapla_metadata.datasets._merge import check_variables_consistency
 from dapla_metadata.datasets._merge import merge_metadata
-from dapla_metadata.datasets.compatibility import is_metadata_in_container_structure
-from dapla_metadata.datasets.compatibility import upgrade_metadata
+from dapla_metadata.datasets._merge import report_metadata_consistency
+from dapla_metadata.datasets.compatibility._utils import (
+    is_metadata_in_container_structure,
+)
+from dapla_metadata.datasets.compatibility.model_backwards_compatibility import (
+    upgrade_metadata,
+)
 from dapla_metadata.datasets.dapla_dataset_path_info import DaplaDatasetPathInfo
 from dapla_metadata.datasets.dataset_parser import DatasetParser
+from dapla_metadata.datasets.dataset_parser import pretty_print_supported_types
 from dapla_metadata.datasets.model_validation import ValidateDatadocMetadata
 from dapla_metadata.datasets.statistic_subject_mapping import StatisticSubjectMapping
 from dapla_metadata.datasets.utility.constants import (
@@ -126,6 +131,7 @@ def __init__(
         self.variables_lookup: dict[str, VariableType] = {}
         self.explicitly_defined_metadata_document = False
         self.dataset_consistency_status: list[DatasetConsistencyStatus] = []
+        self.concrete_data_types_lookup: dict[str, str] = {}
         if metadata_document_path:
             self.metadata_document = UPath(metadata_document_path)
             self.explicitly_defined_metadata_document = True
@@ -169,42 +175,37 @@ def _extract_metadata_from_files(self) -> None:
                 self.metadata_document,
             )
 
-        if (
-            self.dataset_path is not None
-            and self.dataset == all_optional_model.Dataset()
-            and len(self.variables) == 0
-        ):
+        if self.dataset_path:
             extracted_metadata = self._extract_metadata_from_dataset(self.dataset_path)
+            self.dataset_consistency_status.extend(
+                self.check_illegal_variable_data_type(
+                    extracted_metadata.variables or [], self.concrete_data_types_lookup
+                )
+            )
 
         if (
             self.dataset_path
             and self.metadata_document
             and extracted_metadata
             and existing_metadata
-        ):
-            self.dataset_consistency_status = check_dataset_consistency(
-                self.dataset_path,
-                self.metadata_document,
+        ) and self.explicitly_defined_metadata_document:
+            self.dataset_consistency_status.extend(
+                check_dataset_consistency(
+                    self.dataset_path,
+                    self.metadata_document,
+                )
             )
             self.dataset_consistency_status.extend(
                 check_variables_consistency(
                     extracted_metadata.variables or [],
                     existing_metadata.variables or [],
                 )
             )
-
-        if (
-            self.dataset_path
-            and self.explicitly_defined_metadata_document
-            and self.metadata_document is not None
-            and self.metadata_document.exists()
-            and extracted_metadata is not None
-            and existing_metadata is not None
-        ):
-            check_ready_to_merge(
+            report_metadata_consistency(
                 self.dataset_consistency_status,
                 errors_as_warnings=self.errors_as_warnings,
             )
+            # Merge existing metadata with a new dataset
             merged_metadata = merge_metadata(
                 extracted_metadata,
                 existing_metadata,
@@ -215,8 +216,31 @@ def _extract_metadata_from_files(self) -> None:
                 self.dataset_path,
             )
             self._set_metadata(merged_metadata)
-        else:
-            self._set_metadata(existing_metadata or extracted_metadata)
+            return
+
+        report_metadata_consistency(
+            self.dataset_consistency_status,
+            errors_as_warnings=self.errors_as_warnings,
+            message="Problems were detected with the metadata.",
+        )
+        self._set_metadata(existing_metadata or extracted_metadata)
+
+    def check_illegal_variable_data_type(
+        self, variables: VariableListType, concrete_data_types_lookup: dict[str, str]
+    ) -> list[DatasetConsistencyStatus]:
+        """Check whether any of the variable types are unsupported.
+
+        When we encounter a variable which is unsupported, the `DatasetParser` sets the variable `data_type` to `None`.
+        This function detects that situation and creates a friendly error message to inform of the situation.
+        """
+        return [
+            DatasetConsistencyStatus(
+                message=f"Unsupported data type for variable '{v.short_name}' type: '{concrete_data_types_lookup.get(v.short_name, 'unknown')}' from dataset {self.dataset_path}\nPlease change the type of the variable to one of the supported options:\n{pretty_print_supported_types()}",
+                success=False,
+            )
+            for v in variables
+            if v.short_name and not v.data_type
+        ]
 
     def _set_metadata(
         self,
@@ -369,6 +393,14 @@ def _extract_metadata_from_dataset(
             spatial_coverage_description=DEFAULT_SPATIAL_COVERAGE_DESCRIPTION,
         )
         metadata.variables = DatasetParser.for_file(dataset).get_fields()
+        try:
+            self.concrete_data_types_lookup = DatasetParser.for_file(
+                dataset
+            ).get_concrete_data_types()
+        except RuntimeError:
+            logger.exception(
+                "Failed to get concrete data types for dataset %s", dataset
+            )
         return metadata
 
     @staticmethod

src/dapla_metadata/datasets/dataset_parser.py

@@ -9,6 +9,7 @@
 from abc import ABC
 from abc import abstractmethod
 from typing import TYPE_CHECKING
+from typing import ClassVar
 
 import pandas as pd
 from datadoc_model.all_optional.model import DataType
@@ -24,6 +25,11 @@
     import pyarrow as pa
     from upath.types import ReadablePathLike
 
+
+PARQUET_FILE_SUFFIX = ".parquet"
+PARQUET_GZIP_FILE_SUFFIX = ".parquet.gzip"
+SAS7BDAT_FILE_SUFFIX = ".sas7bdat"
+
 KNOWN_INTEGER_TYPES = (
     "int",
     "int_",
@@ -98,6 +104,11 @@
     TYPE_MAP.update(dict.fromkeys(concrete_type, abstract_type))
 
 
+def pretty_print_supported_types() -> str:
+    """Return a human-readable string of the supported data types."""
+    return "\n".join(f"{t[1].value}: {t[0]}" for t in TYPE_CORRESPONDENCE)
+
+
 class DatasetParser(ABC):
     """Abstract Base Class for all Dataset parsers.
 
@@ -162,12 +173,18 @@ def transform_data_type(data_type: str) -> DataType | None:
 
     @abstractmethod
     def get_fields(self) -> list[Variable]:
-        """Abstract method, must be implemented by subclasses."""
+        """Extract the variable names and abstract data types for this dataset."""
+
+    @abstractmethod
+    def get_concrete_data_types(self) -> dict[str, str]:
+        """Extract the variable names and concrete data types for this dataset."""
 
 
 class DatasetParserParquet(DatasetParser):
     """Concrete implementation for parsing parquet files."""
 
+    _EXCLUDED_VARIABLE_NAMES: ClassVar[set[str]] = {"__index_level_0__"}
+
     def __init__(self, dataset: UPath) -> None:
         """Call the super init method for initialization.
 
@@ -178,18 +195,24 @@ def __init__(self, dataset: UPath) -> None:
 
     def get_fields(self) -> list[Variable]:
         """Extract the fields from this dataset."""
-        with self.dataset.open(mode="rb") as f:
-            schema: pa.Schema = pq.read_schema(f)  # type: ignore [arg-type, assignment]
         return [
             Variable(
-                short_name=data_field.name.strip(),
-                data_type=self.transform_data_type(str(data_field.type)),  # type: ignore [attr-defined]
+                short_name=data_field[0],
+                data_type=self.transform_data_type(data_field[1]),
             )
-            for data_field in schema
-            if data_field.name
-            != "__index_level_0__"  # Index columns should not be documented
+            for data_field in self.get_concrete_data_types().items()
         ]
 
+    def get_concrete_data_types(self) -> dict[str, str]:
+        """Extract the variable names and concrete data types for this dataset."""
+        with self.dataset.open(mode="rb") as f:
+            schema: pa.Schema = pq.read_schema(f)
+        return {
+            data_field.name.strip(): str(data_field.type)
+            for data_field in schema
+            if data_field.name not in self._EXCLUDED_VARIABLE_NAMES
+        }
+
 
 class DatasetParserSas7Bdat(DatasetParser):
     """Concrete implementation for parsing SAS7BDAT files."""
@@ -240,10 +263,10 @@ def get_fields(self) -> list[Variable]:
 
         return fields
 
+    def get_concrete_data_types(self) -> dict[str, str]:
+        """Extract the variable names and concrete data types for this dataset."""
+        raise NotImplementedError
 
-PARQUET_FILE_SUFFIX = ".parquet"
-PARQUET_GZIP_FILE_SUFFIX = ".parquet.gzip"
-SAS7BDAT_FILE_SUFFIX = ".sas7bdat"
 
 SUPPORTED_DATASET_FILE_SUFFIXES: dict[
     str,

tests/datasets/resources/datasets/category_data_type.parquet

@@ -25,6 +25,7 @@
 from pydantic import ValidationError
 
 from dapla_metadata.dapla.user_info import TestUserInfo
+from dapla_metadata.datasets._merge import InconsistentDatasetsError
 from dapla_metadata.datasets.core import Datadoc
 from dapla_metadata.datasets.statistic_subject_mapping import StatisticSubjectMapping
 from dapla_metadata.datasets.utility.constants import (
@@ -602,3 +603,12 @@ def test_merge_with_fewer_variables_in_existing_metadata(tmp_path):
         "bankinnskudd",
         "dato",
     ]
+
+
+def test_unknown_data_type():
+    with pytest.raises(
+        InconsistentDatasetsError, match="Unsupported data type for variable"
+    ):
+        Datadoc(
+            dataset_path="tests/datasets/resources/datasets/category_data_type.parquet"
+        )

tests/datasets/test_datadoc_metadata.py

@@ -22,8 +22,8 @@
 from dapla_metadata.datasets._merge import InconsistentDatasetsError
 from dapla_metadata.datasets._merge import InconsistentDatasetsWarning
 from dapla_metadata.datasets._merge import check_dataset_consistency
-from dapla_metadata.datasets._merge import check_ready_to_merge
 from dapla_metadata.datasets._merge import check_variables_consistency
+from dapla_metadata.datasets._merge import report_metadata_consistency
 from dapla_metadata.datasets.core import Datadoc
 from tests.datasets.constants import TEST_BUCKET_NAMING_STANDARD_COMPATIBLE_PATH
 from tests.datasets.constants import VARIABLE_DATA_TYPES
@@ -140,7 +140,7 @@ def test_check_dataset_consistency_inconsistent_paths(
     [True, False],
     ids=["warnings", "errors"],
 )
-def test_check_ready_to_merge_errors_as_warnings(
+def test_report_metadata_consistency_errors_as_warnings(
     dataset_consistency_status: list[DatasetConsistencyStatus],
     errors_as_warnings: bool,
 ):
@@ -149,7 +149,7 @@ def test_check_ready_to_merge_errors_as_warnings(
             stack.enter_context(pytest.warns(InconsistentDatasetsWarning))
         else:
             stack.enter_context(pytest.raises(InconsistentDatasetsError))
-        check_ready_to_merge(
+        report_metadata_consistency(
             dataset_consistency_status,
             errors_as_warnings=errors_as_warnings,
         )

tests/datasets/test_dataset_consistency.py

@@ -49,6 +49,28 @@ def test_get_fields_parquet(local_parser: DatasetParserParquet):
     assert local_parser.get_fields() == expected_fields
 
 
+@pytest.mark.parametrize(
+    "local_parser",
+    [
+        DatasetParser.for_file(TEST_PARQUET_FILEPATH),
+        DatasetParser.for_file(TEST_PARQUET_GZIP_FILEPATH),
+    ],
+)
+def test_get_concrete_data_types_parquet(local_parser: DatasetParserParquet):
+    expected_fields = {
+        "alm_inntekt": "int64",
+        "ber_bruttoformue": "int64",
+        "fullf_utdanning": "string",
+        "hoveddiagnose": "string",
+        "pers_id": "string",
+        "sivilstand": "string",
+        "sykepenger": "int64",
+        "tidspunkt": "timestamp[us]",
+    }
+
+    assert local_parser.get_concrete_data_types() == expected_fields
+
+
 def test_get_fields_sas7bdat():
     expected_fields = [
         Variable(
@@ -101,8 +123,7 @@ def test_transform_datatype_unknown_type():
     ],
 )
 def test_transform_datatype(expected: DataType, concrete_type: str):
-    actual = DatasetParser.transform_data_type(concrete_type)
-    assert actual == expected
+    assert DatasetParser.transform_data_type(concrete_type) == expected
 
 
 @pytest.fixture