Add parsing/support for UDP links

Author: soxofaan

src/esa_apex_toolbox/algorithms.py

@@ -8,6 +8,10 @@
 import requests
 
 
+class LINK_REL:
+    UDP = "udp"
+
+
 def _load_json_resource(src: Union[dict, str, Path]) -> dict:
     """Load a JSON resource from a file or a string."""
     if isinstance(src, dict):
@@ -32,11 +36,38 @@ def _load_json_resource(src: Union[dict, str, Path]) -> dict:
         raise ValueError(f"Unsupported JSON resource type {type(src)}")
 
 
+class InvalidMetadataError(ValueError):
+    pass
+
+
+@dataclasses.dataclass(frozen=True)
+class UdpLink:
+    href: str
+    title: Optional[str] = None
+
+    @classmethod
+    def from_link_object(cls, data: dict) -> UdpLink:
+        """Parse a link object (dict/mapping) into a UdpLink object."""
+        if "rel" not in data:
+            raise InvalidMetadataError("Missing 'rel' attribute in link object")
+        if data["rel"] != LINK_REL.UDP:
+            raise InvalidMetadataError(f"Expected link with rel='udp' but got {data['rel']!r}")
+        if "type" in data and data["type"] != "application/json":
+            raise InvalidMetadataError(f"Expected link with type='application/json' but got {data['type']!r}")
+        if "href" not in data:
+            raise InvalidMetadataError("Missing 'href' attribute in link object")
+        return cls(
+            href=data["href"],
+            title=data.get("title"),
+        )
+
+
 @dataclasses.dataclass(frozen=True)
 class Algorithm:
     id: str
     title: Optional[str] = None
     description: Optional[str] = None
+    udp_link: Optional[UdpLink] = None
     # TODO more fields
 
     @classmethod
@@ -47,18 +78,27 @@ def from_ogc_api_record(cls, src: Union[dict, str, Path]) -> Algorithm:
         """
         data = _load_json_resource(src)
 
-        # TODO dedicated exceptions for structure/schema violations
         if not data.get("type") == "Feature":
-            raise ValueError("Expected a GeoJSON Feature object")
+            raise InvalidMetadataError(f"Expected a GeoJSON 'Feature' object, but got type {data.get('type')!r}.")
         if "http://www.opengis.net/spec/ogcapi-records-1/1.0/req/record-core" not in data.get("conformsTo", []):
-            raise ValueError("Expected an OGC API - Records record object")
+            raise InvalidMetadataError(
+                f"Expected an 'OGC API - Records' record object, but got {data.get('conformsTo')!r}."
+            )
 
         properties = data.get("properties", {})
-        if not properties.get("type") == "apex_algorithm":
-            raise ValueError("Expected an APEX algorithm object")
+        if properties.get("type") != "apex_algorithm":
+            raise InvalidMetadataError(f"Expected an APEX algorithm object, but got type {properties.get('type')!r}.")
+
+        links = data.get("links", [])
+        udp_links = [UdpLink.from_link_object(link) for link in links if link.get("rel") == LINK_REL.UDP]
+        if len(udp_links) > 1:
+            raise InvalidMetadataError("Multiple UDP links found")
+        # TODO: is having a UDP link a requirement?
+        udp_link = udp_links[0] if udp_links else None
 
         return cls(
             id=data["id"],
             title=properties.get("title"),
             description=properties.get("description"),
+            udp_link=udp_link,
         )

tests/data/ogcapi-records/algorithm01.json

@@ -13,7 +13,7 @@
        {
         "rel": "udp",
         "type": "application/json",
-        "title": "openEO UDP",
+        "title": "UDP One",
         "href": "https://esa-apex.test/udp/algorithm01.json"
       }
     ]

tests/test_algorithms.py

@@ -1,13 +1,116 @@
+import re
 from pathlib import Path
 
 import pytest
 
-from esa_apex_toolbox.algorithms import Algorithm
+from esa_apex_toolbox.algorithms import Algorithm, InvalidMetadataError, UdpLink
 
 DATA_ROOT = Path(__file__).parent / "data"
 
 
+class TestUdpLink:
+    def test_from_link_object_basic(self):
+        data = {
+            "rel": "udp",
+            "href": "https://esa-apex.test/udp/basic.json",
+        }
+        link = UdpLink.from_link_object(data)
+        assert link.href == "https://esa-apex.test/udp/basic.json"
+        assert link.title is None
+
+    def test_from_link_object_with_title(self):
+        data = {
+            "rel": "udp",
+            "href": "https://esa-apex.test/udp/basic.json",
+            "title": "My basic UDP",
+        }
+        link = UdpLink.from_link_object(data)
+        assert link.href == "https://esa-apex.test/udp/basic.json"
+        assert link.title == "My basic UDP"
+
+    def test_from_link_object_missing_rel(self):
+        data = {
+            "href": "https://esa-apex.test/udp/basic.json",
+        }
+        with pytest.raises(InvalidMetadataError, match="Missing 'rel' attribute"):
+            _ = UdpLink.from_link_object(data)
+
+    def test_from_link_object_wrong_rel(self):
+        data = {
+            "rel": "self",
+            "href": "https://esa-apex.test/udp/basic.json",
+        }
+        with pytest.raises(InvalidMetadataError, match="Expected link with rel='udp'"):
+            _ = UdpLink.from_link_object(data)
+
+    def test_from_link_object_no_href(self):
+        data = {
+            "rel": "udp",
+        }
+        with pytest.raises(InvalidMetadataError, match="Missing 'href' attribute"):
+            _ = UdpLink.from_link_object(data)
+
+    def test_from_link_object_wrong_type(self):
+        data = {
+            "rel": "udp",
+            "href": "https://esa-apex.test/udp/basic.json",
+            "type": "application/xml",
+        }
+        with pytest.raises(InvalidMetadataError, match="Expected link with type='application/json'"):
+            _ = UdpLink.from_link_object(data)
+
+
 class TestAlgorithm:
+    def test_from_ogc_api_record_minimal(self):
+        data = {
+            "id": "minimal",
+            "type": "Feature",
+            "conformsTo": ["http://www.opengis.net/spec/ogcapi-records-1/1.0/req/record-core"],
+            "properties": {
+                "type": "apex_algorithm",
+            },
+        }
+        algorithm = Algorithm.from_ogc_api_record(data)
+        assert algorithm.id == "minimal"
+
+    def test_from_ogc_api_record_wrong_type(self):
+        data = {
+            "id": "wrong",
+            "type": "apex_algorithm",
+            "conformsTo": ["http://www.opengis.net/spec/ogcapi-records-1/1.0/req/record-core"],
+        }
+        with pytest.raises(
+            InvalidMetadataError, match="Expected a GeoJSON 'Feature' object, but got type 'apex_algorithm'."
+        ):
+            _ = Algorithm.from_ogc_api_record(data)
+
+    def test_from_ogc_api_record_wrong_conform(self):
+        data = {
+            "id": "wrong",
+            "type": "Feature",
+            "conformsTo": ["http://nope.test/"],
+            "properties": {
+                "type": "apex_algorithm",
+            },
+        }
+        with pytest.raises(
+            InvalidMetadataError,
+            match=re.escape("Expected an 'OGC API - Records' record object, but got ['http://nope.test/']."),
+        ):
+            _ = Algorithm.from_ogc_api_record(data)
+
+    def test_from_ogc_api_record_wrong_properties_type(self):
+        data = {
+            "id": "wrong",
+            "type": "Feature",
+            "conformsTo": ["http://www.opengis.net/spec/ogcapi-records-1/1.0/req/record-core"],
+            "properties": {
+                "type": "udp",
+            },
+        }
+        with pytest.raises(InvalidMetadataError, match="Expected an APEX algorithm object, but got type 'udp'"):
+            _ = Algorithm.from_ogc_api_record(data)
+
     def test_from_ogc_api_record_basic(self):
         data = {
             "id": "basic",
@@ -22,7 +125,7 @@ def test_from_ogc_api_record_basic(self):
                 {
                     "rel": "udp",
                     "type": "application/json",
-                    "title": "openEO UDP",
+                    "title": "Basic UDP",
                     "href": "https://esa-apex.test/udp/basic.json",
                 }
             ],
@@ -31,6 +134,10 @@ def test_from_ogc_api_record_basic(self):
         assert algorithm.id == "basic"
         assert algorithm.title == "Basic"
         assert algorithm.description == "The basics."
+        assert algorithm.udp_link == UdpLink(
+            href="https://esa-apex.test/udp/basic.json",
+            title="Basic UDP",
+        )
 
     @pytest.mark.parametrize("path_type", [str, Path])
     def test_from_ogc_api_record_path(self, path_type):
@@ -39,13 +146,21 @@ def test_from_ogc_api_record_path(self, path_type):
         assert algorithm.id == "algorithm01"
         assert algorithm.title == "Algorithm One"
         assert algorithm.description == "A first algorithm."
+        assert algorithm.udp_link == UdpLink(
+            href="https://esa-apex.test/udp/algorithm01.json",
+            title="UDP One",
+        )
 
     def test_from_ogc_api_record_str(self):
         dump = (DATA_ROOT / "ogcapi-records/algorithm01.json").read_text()
         algorithm = Algorithm.from_ogc_api_record(dump)
         assert algorithm.id == "algorithm01"
         assert algorithm.title == "Algorithm One"
         assert algorithm.description == "A first algorithm."
+        assert algorithm.udp_link == UdpLink(
+            href="https://esa-apex.test/udp/algorithm01.json",
+            title="UDP One",
+        )
 
     def test_from_ogc_api_record_url(self, requests_mock):
         url = "https://esa-apex.test/algorithms/a1.json"
@@ -55,3 +170,7 @@ def test_from_ogc_api_record_url(self, requests_mock):
         assert algorithm.id == "algorithm01"
         assert algorithm.title == "Algorithm One"
         assert algorithm.description == "A first algorithm."
+        assert algorithm.udp_link == UdpLink(
+            href="https://esa-apex.test/udp/algorithm01.json",
+            title="UDP One",
+        )