WIP: Enrich UserDefinedProcesses API

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

Author: soxofaan

openeo_driver/backend.py

@@ -634,6 +634,47 @@ def from_dict(cls, d: dict) -> 'UserDefinedProcessMetadata':
     def prepare_for_json(self) -> dict:
         return self._asdict()  # pylint: disable=no-member
 
+    def to_api_dict(self, full=True, user: User = None) -> dict:
+        """API-version-aware conversion of UDP metadata to jsonable dict"""
+        d = self.prepare_for_json()
+        if not full:
+            # API recommends to limit response size by omitting larger/optional fields
+            d = {k: v for k, v in d.items() if k in ["id", "summary", "description", "parameters", "returns"]}
+        elif self.public and user:
+            namespace = "u:" + user.user_id
+            d["links"] = (d.get("links") or []) + [
+                {
+                    "rel": "canonical",
+                    # TODO: use signed url?
+                    "href": flask.url_for(
+                        ".processes_details", namespace=namespace, process_id=self.id, _external=True
+                    ),
+                    "title": f"Public URL for user-defined process {self.id!r}",
+                }
+            ]
+        elif "public" in d and not d["public"]:
+            # Don't include non-standard "public" field when false to stay closer to standard API
+            del d["public"]
+
+        return dict_no_none(**d)
+
+
+class UserDefinedProcessesListing:
+    def __init__(self, udps: List[UserDefinedProcessMetadata]):
+        self._udps = udps
+
+    def to_response_dict(self, build_url: Callable[[dict], str], api_version: ComparableVersion = None) -> dict:
+        """
+        Produce `GET /process_graphs` response data, to be JSONified.
+
+        :param build_url: function to generate a paginated" URL from given pagination related parameters,
+            e.g. `lambda params: flask.url_for(".list_jobs", **params, _external=True)`
+        """
+        return {
+            "processes": [udp.to_api_dict(full=False) for udp in self._udps],
+            "links": [],
+        }
+
 
 class UserDefinedProcesses(MicroService):
     """
@@ -653,8 +694,19 @@ 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 remove this deprecated API when unused
         raise NotImplementedError
 
+    def list_for_user(
+        self,
+        user_id: str,
+        # TODO: encapsulate limit and request_parameters in a pagination object?
+        limit: Optional[int] = None,
+        request_parameters: Optional[dict] = None,
+    ) -> UserDefinedProcessesListing:
+        udps = self.get_for_user(user_id=user_id)
+        ...
+
     def save(self, user_id: str, process_id: str, spec: dict) -> None:
         """
         Store new UDP

openeo_driver/views.py

@@ -761,7 +761,7 @@ def processes_from_namespace(namespace):
         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]
-            processes = [_jsonable_udp_metadata(udp, full=full, user=User(user_id=user_id)) for udp in user_udps]
+            processes = [udp.to_api_dict(full=full, user=User(user_id=user_id)) for udp in user_udps]
         elif ":" not in namespace:
             process_registry = backend_implementation.processing.get_process_registry(
                 api_version=requested_api_version()
@@ -796,7 +796,7 @@ def processes_details(namespace, process_id):
             udp = backend_implementation.user_defined_processes.get(user_id=user_id, process_id=process_id)
             if not udp:
                 raise ProcessUnsupportedException(process=process_id, namespace=namespace)
-            process = _jsonable_udp_metadata(udp, full=True, user=User(user_id=user_id))
+            process = udp.to_api_dict(full=True, user=User(user_id=user_id))
         elif ":" not in namespace:
             process_registry = backend_implementation.processing.get_process_registry(
                 api_version=requested_api_version()
@@ -1754,21 +1754,17 @@ def udp_get(process_graph_id: str, user: User):
         _check_valid_process_graph_id(process_id=process_graph_id)
         udp = backend_implementation.user_defined_processes.get(user_id=user.user_id, process_id=process_graph_id)
         if udp:
-            return _jsonable_udp_metadata(udp, full=True , user=user)
+            return udp.to_api_dict(full=True, user=user)
 
         raise ProcessGraphNotFoundException(process_graph_id)
 
     @api_endpoint
     @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': [_jsonable_udp_metadata(udp, full=False) for udp in user_udps],
-            # TODO: pagination links?
-            # TODO: allow backend_implementation to define links?
-            "links": [],
-        }
+        # TODO #377 pagination support
+        udps = backend_implementation.user_defined_processes.list_for_user(user_id=user.user_id)
+        return jsonify(udps.to_response_dict())
 
     @api_endpoint
     @blueprint.route('/process_graphs/<process_graph_id>', methods=['DELETE'])
@@ -1779,27 +1775,6 @@ def udp_delete(process_graph_id: str, user: User):
         return response_204_no_content()
 
 
-def _jsonable_udp_metadata(metadata: UserDefinedProcessMetadata, full=True, user: User = None) -> dict:
-    """API-version-aware conversion of UDP metadata to jsonable dict"""
-    d = metadata.prepare_for_json()
-    if not full:
-        # API recommends to limit response size by omitting larger/optional fields
-        d = {k: v for k, v in d.items() if k in ["id", "summary", "description", "parameters", "returns"]}
-    elif metadata.public and user:
-        namespace = "u:" + user.user_id
-        d["links"] = (d.get("links") or []) + [
-            {
-                "rel": "canonical",
-                # TODO: use signed url?
-                "href": url_for(".processes_details", namespace=namespace, process_id=metadata.id, _external=True),
-                "title": f"Public URL for user-defined process {metadata.id!r}"
-            }
-        ]
-    elif "public" in d and not d["public"]:
-        # Don't include non-standard "public" field when false to stay closer to standard API
-        del d["public"]
-
-    return dict_no_none(**d)
 
 
 def _normalize_collection_metadata(metadata: dict, api_version: ComparableVersion, full=False) -> dict: