Issue #125 Add "federation:missing" to /process_graphs

Author: soxofaan

CHANGELOG.md

@@ -11,6 +11,7 @@ The format is roughly based on [Keep a Changelog](https://keepachangelog.com/en/
 - Bump minimum required Python version to 3.11 ([#127](https://github.com/Open-EO/openeo-aggregator/issues/127), [#174](https://github.com/Open-EO/openeo-aggregator/issues/174))
 - Add title, description, online/offline status and last_status_check to federation listing in capabilities document ([#22](https://github.com/Open-EO/openeo-aggregator/issues/22))
 - Provide "federation:missing" on `/file_formats` ([#124](https://github.com/Open-EO/openeo-aggregator/issues/124))
+- Provide "federation:missing" on `/process_graphs` ([#125](https://github.com/Open-EO/openeo-aggregator/issues/125))
 
 ## 0.43.0
 

setup.py

@@ -35,7 +35,7 @@
         "requests",
         "attrs",
         "openeo>=0.38.0",
-        "openeo_driver>=0.126.1.dev",
+        "openeo_driver>=0.130.0.dev",
         "flask~=2.0",
         "gunicorn~=20.0",
         "python-json-logger>=2.0.0",

src/openeo_aggregator/backend.py

@@ -51,6 +51,7 @@
     SecondaryServices,
     ServiceMetadata,
     UserDefinedProcesses,
+    UserDefinedProcessesListing,
     UserDefinedProcessMetadata,
 )
 from openeo_driver.datacube import DriverDataCube
@@ -1392,17 +1393,31 @@ def update_service(self, user_id: str, service_id: str, process_graph: dict) ->
                 ) from e
 
 
+class AggregatorUserDefinedProcessesListing(UserDefinedProcessesListing):
+
+    def __init__(self, udps: List[UserDefinedProcessMetadata], *, federation_missing: Iterable[str] = ()):
+        super().__init__(udps=udps)
+        self._federation_missing = list(federation_missing)
+
+    def to_response_dict(self, full: bool = True) -> dict:
+        resp = super().to_response_dict(full=full)
+        resp["federation:missing"] = self._federation_missing
+        return resp
+
+
 class AggregatorUserDefinedProcesses(UserDefinedProcesses):
     def __init__(self, backends: MultiBackendConnection):
         super(AggregatorUserDefinedProcesses, self).__init__()
         self._backends = backends
 
     @contextlib.contextmanager
-    def _get_connection(self, process_graph_id: Optional[str] = None) -> Iterator[openeo.Connection]:
+    def _get_connection(self, process_graph_id: Optional[str] = None) -> Iterator[BackendConnection]:
         """Get connection and handle/translate common errors"""
         try:
-            # TODO: we blindly pick "first" upstream backend for now. Do better!
-            with self._backends.first().authenticated_from_request(request=flask.request) as con:
+            # TODO #90 #176 we blindly pick "first" upstream backend for now. Do better!
+            first = self._backends.first()
+            _log.warning(f"AggregatorUserDefinedProcesses: just using 'first' backend: {first!r}")
+            with first.authenticated_from_request(request=flask.request) as con:
                 yield con
         except OpenEoApiError as e:
             if e.code == ProcessGraphNotFoundException.code:
@@ -1414,10 +1429,17 @@ def get(self, user_id: str, process_id: str) -> Union[UserDefinedProcessMetadata
             metadata = con.get(f"/process_graphs/{process_id}", expected_status=200).json()
             return UserDefinedProcessMetadata.from_dict(metadata)
 
-    def get_for_user(self, user_id: str) -> List[UserDefinedProcessMetadata]:
+    def list_for_user(
+        self,
+        user_id: str,
+    ) -> UserDefinedProcessesListing:
+        # TODO #175 how to handle pagination?
         with self._get_connection() as con:
             data = con.get(f"/process_graphs", expected_status=200).json()
-            return [UserDefinedProcessMetadata.from_dict(p) for p in data["processes"]]
+            udps = [UserDefinedProcessMetadata.from_dict(p) for p in data["processes"]]
+            federation_missing = [c for c in self._backends.get_all_connection_ids() if c != con.id]
+
+        return AggregatorUserDefinedProcessesListing(udps=udps, federation_missing=federation_missing)
 
     def save(self, user_id: str, process_id: str, spec: dict) -> None:
         with self._get_connection(process_graph_id=process_id) as con:

src/openeo_aggregator/connection.py

@@ -303,6 +303,10 @@ def get_connections(self) -> List[BackendConnection]:
     def __iter__(self) -> Iterator[BackendConnection]:
         return iter(self.get_connections())
 
+    def get_all_connection_ids(self) -> List[str]:
+        """Return all connection ids (regardless of online/offline status)."""
+        return list(self._backend_urls.keys())
+
     def get_disabled_connection_ids(self) -> Set[str]:
         all_ids = set(self._backend_urls.keys())
         active_ids = set(b.id for b in self.get_connections())

tests/test_connection.py

@@ -674,10 +674,12 @@ def test_get_connections(self, requests_mock, backend1, backend2):
 
         assert set(b.id for b in multi_backend_connection.get_connections()) == {"b1", "b2"}
         assert multi_backend_connection.get_disabled_connection_ids() == set()
+        assert multi_backend_connection.get_all_connection_ids() == ["b1", "b2"]
 
         # Wait for connections cache to expire
         with clock_mock(offset=1000):
             requests_mock.get(backend1 + "/", status_code=500, json={"error": "nope"})
 
             assert set(b.id for b in multi_backend_connection.get_connections()) == {"b2"}
             assert multi_backend_connection.get_disabled_connection_ids() == {"b1"}
+            assert multi_backend_connection.get_all_connection_ids() == ["b1", "b2"]

tests/test_views.py

@@ -3017,7 +3017,11 @@ def test_list_udps_empty(self, api100, requests_mock, backend1):
         )
         api100.set_auth_bearer_token(token=TEST_USER_BEARER_TOKEN)
         res = api100.get("/process_graphs").assert_status_code(200).json
-        assert res == {"processes": [], "links": []}
+        assert res == {
+            "processes": [],
+            "links": [],
+            "federation:missing": ["b2"],
+        }
         assert upstream.call_count == 1
 
     def test_list_udps_existing(self, api100, requests_mock, backend1):
@@ -3043,6 +3047,7 @@ def test_list_udps_existing(self, api100, requests_mock, backend1):
                 {"id": "somethingelse"},
             ],
             "links": [],
+            "federation:missing": ["b2"],
         }
         assert upstream.call_count == 1