Move file-level error reporting for duplicate keys to read_json

Author: duckontheweb

pystac/stac_io.py

@@ -3,7 +3,6 @@
 import json
 from typing import (
     Any,
-    Callable,
     Dict,
     List,
     Optional,
@@ -18,7 +17,7 @@
 from urllib.error import HTTPError
 
 import pystac
-from pystac.utils import safe_urlparse, get_opt
+from pystac.utils import safe_urlparse
 from pystac.serialization import (
     merge_common_properties,
     identify_stac_object_type,
@@ -105,8 +104,7 @@ def json_dumps(self, json_dict: Dict[str, Any], *args: Any, **kwargs: Any) -> st
         This method may be overwritten in :class:`StacIO` sub-classes to provide custom
         serialization logic. The method accepts arbitrary keyword arguments. These are
         not used by the default implementation, but may be used by sub-class
-        implementations (see :meth:`DuplicateKeyReportingMixin.json_dumps` as an
-        example).
+        implementations.
 
         Args:
 
@@ -300,36 +298,51 @@ class DuplicateKeyReportingMixin(StacIO):
     See https://github.com/stac-utils/pystac/issues/313
     """
 
-    def json_loads(self, txt: str, *args: Any, **kwargs: Any) -> Dict[str, Any]:
-        source: Union[str, "Link_Type"] = get_opt(kwargs.get("source"))
+    def json_loads(self, txt: str, *_: Any, **__: Any) -> Dict[str, Any]:
+        """Overwrites :meth:`StacIO.json_loads` as the internal method used by
+        :class:`DuplicateKeyReportingMixin` for deserializing a JSON string to a
+        dictionary while checking for duplicate object keys.
+
+        Raises:
+
+            pystac.DuplicateObjectKeyError : If a duplicate object key is found.
+        """
         result: Dict[str, Any] = json.loads(
-            txt, object_pairs_hook=self.duplicate_object_names_report_builder(source)
+            txt, object_pairs_hook=self._report_duplicate_object_names
         )
         return result
 
-    @staticmethod
-    def duplicate_object_names_report_builder(
-        source: Union[str, "Link_Type"]
-    ) -> Callable[[List[Tuple[str, Any]]], Dict[str, Any]]:
-        def report_duplicate_object_names(
-            object_pairs: List[Tuple[str, Any]]
-        ) -> Dict[str, Any]:
-            result: Dict[str, Any] = {}
-            for key, value in object_pairs:
-                if key in result:
-                    url = (
-                        source
-                        if isinstance(source, str)
-                        else source.get_absolute_href()
-                    )
-                    raise DuplicateObjectKeyError(
-                        f"Found duplicate object name “{key}” in “{url}”"
-                    )
-                else:
-                    result[key] = value
-            return result
+    def read_json(
+        self, source: Union[str, "Link_Type"], *args: Any, **kwargs: Any
+    ) -> Dict[str, Any]:
+        """Overwrites :meth:`StacIO.read_json` for deserializing a JSON file to a
+        dictionary while checking for duplicate object keys.
+
+        Raises:
+
+            pystac.DuplicateObjectKeyError : If a duplicate object key is found.
+        """
+        txt = self.read_text(source, *args, **kwargs)
+        try:
+            return self.json_loads(txt, source=source)
+        except pystac.DuplicateObjectKeyError as e:
+            url = source if isinstance(source, str) else source.get_absolute_href()
+            msg = str(e) + f" in {url}"
+            raise pystac.DuplicateObjectKeyError(msg)
 
-        return report_duplicate_object_names
+    @staticmethod
+    def _report_duplicate_object_names(
+        object_pairs: List[Tuple[str, Any]]
+    ) -> Dict[str, Any]:
+        result: Dict[str, Any] = {}
+        for key, value in object_pairs:
+            if key in result:
+                raise pystac.DuplicateObjectKeyError(
+                    f'Found duplicate object name "{key}"'
+                )
+            else:
+                result[key] = value
+        return result
 
 
 class STAC_IO:

tests/test_stac_io.py

@@ -4,7 +4,7 @@
 import tempfile
 
 import pystac
-from pystac.stac_io import STAC_IO, StacIO
+from pystac.stac_io import STAC_IO, StacIO, DefaultStacIO, DuplicateKeyReportingMixin
 from tests.utils import TestCases
 
 
@@ -117,3 +117,31 @@ def test_read_from_stac_object(self) -> None:
             TestCases.get_path("data-files/catalogs/test-case-1/catalog.json")
         )
         self.assertIsInstance(catalog, pystac.Catalog)
+
+    def test_report_duplicate_keys(self) -> None:
+        # Directly from dict
+        class ReportingStacIO(DefaultStacIO, DuplicateKeyReportingMixin):
+            pass
+
+        stac_io = ReportingStacIO()
+        test_json = """{
+            "key": "value_1",
+            "key": "value_2"
+        }"""
+
+        with self.assertRaises(pystac.DuplicateObjectKeyError) as excinfo:
+            stac_io.json_loads(test_json)
+        self.assertEqual(str(excinfo.exception), 'Found duplicate object name "key"')
+
+        # From file
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            src_href = os.path.join(tmp_dir, "test.json")
+            with open(src_href, "w") as dst:
+                dst.write(test_json)
+
+            with self.assertRaises(pystac.DuplicateObjectKeyError) as excinfo:
+                stac_io.read_json(src_href)
+            self.assertEqual(
+                str(excinfo.exception),
+                f'Found duplicate object name "key" in {src_href}',
+            )