Allow customization of GET /process_graphs

for Open-EO/openeo-aggregator#125
and related to #377

Author: soxofaan

openeo_driver/backend.py

@@ -653,6 +653,18 @@ def to_api_dict(self, *, full: bool = True) -> dict:
         return dict_no_none(**d)
 
 
+class UserDefinedProcessesListing:
+    # TODO #377 pagination support
+    def __init__(self, udps: List[UserDefinedProcessMetadata]):
+        self.udps = udps
+
+    def to_api_dict(self, full: bool = True) -> dict:
+        return {
+            "processes": [udp.to_api_dict(full=full) for udp in self.udps],
+            "links": [],
+        }
+
+
 class UserDefinedProcesses(MicroService):
     """
     Base contract/implementation for User-Defined Processes "microservice"
@@ -671,8 +683,15 @@ def get_for_user(self, user_id: str) -> List[UserDefinedProcessMetadata]:
         List user's UDPs
         https://openeo.org/documentation/1.0/developers/api/reference.html#operation/list-custom-processes
         """
+        # TODO: eliminate this legacy API method
         raise NotImplementedError
 
+    def list_for_user(self, user_id: str) -> UserDefinedProcessesListing:
+        # TODO: replace this adapter implementation with `raise NotImplementedError`
+        #       once `get_for_user` can be eliminated
+        udps = self.get_for_user(user_id=user_id)
+        return UserDefinedProcessesListing(udps=udps)
+
     def save(self, user_id: str, process_id: str, spec: dict) -> None:
         """
         Store new UDP

openeo_driver/views.py

@@ -761,7 +761,9 @@ def processes_from_namespace(namespace):
         # TODO: convention for user namespace? use '@' instead of "u:"
         if namespace.startswith("u:") and backend_implementation.user_defined_processes:
             user_id = namespace.partition("u:")[-1]
-            user_udps = [p for p in backend_implementation.user_defined_processes.get_for_user(user_id) if p.public]
+            user_udps = [
+                p for p in backend_implementation.user_defined_processes.list_for_user(user_id).udps if p.public
+            ]
             processes = [udp.to_api_dict(full=full) for udp in user_udps]
         elif ":" not in namespace:
             process_registry = backend_implementation.processing.get_process_registry(
@@ -1767,13 +1769,8 @@ def udp_get(process_graph_id: str, user: User):
     @blueprint.route('/process_graphs', methods=['GET'])
     @auth_handler.requires_bearer_auth
     def udp_list_for_user(user: User):
-        user_udps = backend_implementation.user_defined_processes.get_for_user(user.user_id)
-        return {
-            "processes": [udp.to_api_dict(full=False) for udp in user_udps],
-            # TODO: pagination links?
-            # TODO: allow backend_implementation to define links?
-            "links": [],
-        }
+        resp = backend_implementation.user_defined_processes.list_for_user(user_id=user.user_id)
+        return resp.to_api_dict(full=False)
 
     @api_endpoint
     @blueprint.route('/process_graphs/<process_graph_id>', methods=['DELETE'])