Issue #366 add support for extracting job option defaults from remote process definitions

Author: soxofaan

CHANGELOG.md

@@ -21,8 +21,12 @@ and start a new "In Progress" section above it.
 
 ## In progress
 
+
+## 0.126.0
+
 - Add STAC collections conformance class ([#195](https://github.com/Open-EO/openeo-python-driver/issues/195))
 - update openeo_driver/specs/openeo-api/1.x submodule to tag `1.2.0` ([#195](https://github.com/Open-EO/openeo-python-driver/issues/195))
+- Extract job option defaults from UDPs and remote process descriptions ([#366](https://github.com/Open-EO/openeo-python-driver/issues/366),  based on Process Parameter Extension: [Open-EO/openeo-api#471](https://github.com/Open-EO/openeo-api/pull/471))
 
 
 ## 0.125.0

openeo_driver/ProcessGraphDeserializer.py

@@ -58,7 +58,7 @@
     CollectionNotFoundException, ProcessUnsupportedException,
 )
 from openeo_driver.processes import ProcessRegistry, ProcessSpec, DEFAULT_NAMESPACE, ProcessArgs
-from openeo_driver.processgraph import get_process_definition_from_url
+from openeo_driver.processgraph import get_process_definition_from_url, ProcessDefinition
 from openeo_driver.save_result import (
     JSONResult,
     SaveResult,
@@ -71,6 +71,7 @@
 from openeo_driver.specs import SPECS_ROOT, read_spec
 from openeo_driver.util.date_math import month_shift
 from openeo_driver.util.geometry import geojson_to_geometry, geojson_to_multipolygon, spatial_extent_union, BoundingBox
+from openeo_driver.util.http import is_http_url
 from openeo_driver.util.utm import auto_utm_epsg_for_geometry
 from openeo_driver.utils import EvalEnv, smart_bool
 
@@ -1789,7 +1790,7 @@ def apply_process(process_id: str, args: dict, namespace: Union[str, None], env:
         args = {name: convert_node(expr, env=env) for (name, expr) in sorted(args.items())}
 
     # when all arguments and dependencies are resolved, we can run the process
-    if namespace and any(namespace.startswith(p) for p in ["http://", "https://"]):
+    if is_http_url(namespace):
         if namespace.startswith("http://"):
             _log.warning(f"HTTP protocol for namespace based remote process definitions is discouraged: {namespace!r}")
         # TODO: security aspects: only allow for certain users, only allow whitelisted domains, support content hash verification ...?
@@ -2067,7 +2068,7 @@ def evaluate_process_from_url(process_id: str, namespace: str, args: dict, env:
     :param namespace: URL of process definition
     """
     try:
-        process_definition = get_process_definition_from_url(process_id=process_id, url=namespace)
+        process_definition: ProcessDefinition = get_process_definition_from_url(process_id=process_id, url=namespace)
     except OpenEOApiException:
         raise
     except Exception as e:

openeo_driver/_version.py

@@ -1 +1 @@
-__version__ = "0.125.0a1"
+__version__ = "0.126.0a1"

openeo_driver/processgraph.py

@@ -1,11 +1,33 @@
 import logging
 import requests
-from typing import NamedTuple, List, Optional
+from typing import NamedTuple, List, Optional, Union
 from openeo_driver.errors import OpenEOApiException
+from openeo_driver.util.http import is_http_url
 
 _log = logging.getLogger(__name__)
 
 
+class ProcessGraphFlatDict(dict):
+    """
+    Wrapper for the classic "flat dictionary" representation
+    of an openEO process graph, e.g.
+
+        {
+            "lc1": {"process_id": "load_collection", "arguments": {...
+            "rd1": {"process_id": "reduce_dimension", "arguments": {...
+            ...
+        }
+
+    - To be used as type annotation where one wants to clarify
+      what exact kind of dictionary-based process graph representation is expected.
+    - Implemented as a subclass of `dict` to be directly compatible
+      with existing, legacy code that expects a simple dictionary.
+    """
+
+    # TODO: move this to openeo python client library?
+    pass
+
+
 class ProcessDefinition(NamedTuple):
     """
     Like `UserDefinedProcessMetadata`, but with different defaults
@@ -21,6 +43,11 @@ class ProcessDefinition(NamedTuple):
     # Definition what the process returns
     returns: Optional[dict] = None
 
+    # TODO: official naming of these "processing parameter" related properties is undecided at the moment.
+    #       see https://github.com/Open-EO/openeo-api/pull/471#discussion_r1904253964
+    default_job_options: Optional[dict] = None
+    default_synchronous_options: Optional[dict] = None
+
 
 def get_process_definition_from_url(process_id: str, url: str) -> ProcessDefinition:
     """
@@ -69,9 +96,57 @@ def get_process_definition_from_url(process_id: str, url: str) -> ProcessDefinit
             message=f"No valid process definition for {process_id!r} found at {url!r}.",
         )
 
+    # TODO: official property name for these "processing parameters" is undecided at the moment.
+    #       see https://github.com/Open-EO/openeo-api/pull/471#discussion_r1904253964
+    if "default_job_parameters" in spec:
+        _log.warning("Extracting experimental 'default_job_parameters' from process definition.")
+        default_job_options = spec["default_job_parameters"]
+    else:
+        default_job_options = None
+    if "default_synchronous_parameters" in spec:
+        _log.warning("Extracting experimental 'default_synchronous_parameters' from process definition.")
+        default_synchronous_options = spec["default_synchronous_parameters"]
+    else:
+        default_synchronous_options = None
+
     return ProcessDefinition(
         id=process_id,
         process_graph=spec["process_graph"],
         parameters=spec.get("parameters", []),
         returns=spec.get("returns"),
+        default_job_options=default_job_options,
+        default_synchronous_options=default_synchronous_options,
     )
+
+
+def extract_default_job_options_from_process_graph(
+    process_graph: ProcessGraphFlatDict, processing_mode: str = "batch_job"
+) -> Union[dict, None]:
+    """
+    Extract default job options from a process definitions in process graph.
+    based on "Processing Parameters" extension.
+
+    :param process_graph: process graph in flat graph format
+    :param processing_mode: "batch_job" or "synchronous"
+    """
+
+    job_options = []
+    for node in process_graph.values():
+        namespace = node.get("namespace")
+        process_id = node["process_id"]
+        if is_http_url(namespace):
+            process_definition = get_process_definition_from_url(process_id=process_id, url=namespace)
+            if processing_mode == "batch_job" and process_definition.default_job_options:
+                job_options.append(process_definition.default_job_options)
+            elif processing_mode == "synchronous" and process_definition.default_synchronous_options:
+                job_options.append(process_definition.default_synchronous_options)
+
+    if len(job_options) == 0:
+        return None
+    elif len(job_options) == 1:
+        return job_options[0]
+    else:
+        # TODO: how to combine multiple default for same parameters?
+        raise NotImplementedError(
+            "Merging multiple default job options from different process definitions is not yet implemented."
+        )

openeo_driver/util/http.py

@@ -8,6 +8,14 @@
 from openeo.util import repr_truncate
 
 
+def is_http_url(data: Union[str, None], *, allow_http: bool = True, allow_https: bool = True) -> bool:
+    """Does given string look like a valid HTTP(S) URL?"""
+    if not isinstance(data, str):
+        return False
+    data = data.lower()
+    return (allow_http and data.startswith("http://")) or (allow_https and data.startswith("https://"))
+
+
 def requests_with_retry(
     total: int = 3,  # the number of retries, not attempts (which is retries + 1)
     backoff_factor: float = 1,

openeo_driver/views.py

@@ -26,6 +26,7 @@
 
 from openeo.utils.version import ComparableVersion
 from openeo.util import dict_no_none, deep_get, Rfc3339, TimingLogger
+from openeo_driver.processgraph import extract_default_job_options_from_process_graph, ProcessGraphFlatDict
 from openeo_driver.urlsigning import UrlSigner
 from openeo_driver.backend import (
     ServiceMetadata,
@@ -590,11 +591,12 @@ def me(user: User):
         )
 
 
-def _extract_process_graph(post_data: dict) -> dict:
+def _extract_process_graph(post_data: dict) -> ProcessGraphFlatDict:
     """
     Extract process graph dictionary from POST data
 
     see https://github.com/Open-EO/openeo-api/pull/262
+    :return: process graph in flat graph format
     """
     try:
         if requested_api_version().at_least("1.0.0"):
@@ -607,7 +609,7 @@ def _extract_process_graph(post_data: dict) -> dict:
     if not isinstance(pg, dict):
         # TODO: more validity checks for (flat) process graph?
         raise ProcessGraphInvalidException
-    return pg
+    return ProcessGraphFlatDict(pg)
 
 
 def _extract_job_options(post_data: dict, to_ignore: typing.Container[str]) -> Union[dict, None]:
@@ -874,12 +876,20 @@ def register_views_batch_jobs(
     def create_job(user: User):
         # TODO: wrap this job specification in a 1.0-style ProcessGrahpWithMetadata?
         post_data = request.get_json()
+        process_graph = _extract_process_graph(post_data)
         # TODO: preserve original non-process_graph process fields too?
-        process = {"process_graph": _extract_process_graph(post_data)}
-        # TODO: this "job_options" is not part of official API. See https://github.com/Open-EO/openeo-api/issues/276
+        process = {"process_graph": process_graph}
+
         job_options = _extract_job_options(
             post_data, to_ignore=["process", "process_graph", "title", "description", "plan", "budget", "log_level"]
         )
+        job_option_defaults = extract_default_job_options_from_process_graph(
+            process_graph=process_graph, processing_mode="batch_job"
+        )
+        if job_option_defaults:
+            _log.info(f"Extending {job_options=} with extracted {job_option_defaults=}")
+            job_options = {**job_option_defaults, **(job_options or {})}
+
         metadata = {k: post_data[k] for k in ["title", "description", "plan", "budget"] if k in post_data}
         metadata["log_level"] = post_data.get("log_level", DEFAULT_LOG_LEVEL_PROCESSING)
         job_info = backend_implementation.batch_jobs.create_job(

tests/test_processgraph.py

@@ -1,5 +1,9 @@
 from openeo_driver.errors import OpenEOApiException
-from openeo_driver.processgraph import get_process_definition_from_url
+from openeo_driver.processgraph import (
+    get_process_definition_from_url,
+    extract_default_job_options_from_process_graph,
+    ProcessGraphFlatDict,
+)
 
 import pytest
 
@@ -53,3 +57,45 @@ def test_get_process_definition_from_url_listing(self, requests_mock):
         }
         assert pd.parameters == [{"name": "delta", "schema": {"type": "number", "optional": True, "default": 1}}]
         assert pd.returns == {"schema": {"type": "number"}}
+
+
+def test_extract_default_job_options_from_process_graph(requests_mock):
+    requests_mock.get(
+        "https://share.test/add3.json",
+        json={
+            "id": "add3",
+            "process_graph": {
+                "add": {"process_id": "add", "arguments": {"x": {"from_parameter": "x", "y": 3}, "result": True}}
+            },
+            "parameters": [
+                {"name": "x", "schema": {"type": "number"}},
+            ],
+            "returns": {"schema": {"type": "number"}},
+            "default_job_parameters": {
+                "memory": "2GB",
+                "cpu": "yellow",
+            },
+            "default_synchronous_parameters": {
+                "cpu": "green",
+            },
+        },
+    )
+
+    pg = ProcessGraphFlatDict(
+        {
+            "add12": {"process_id": "add", "arguments": {"x": 1, "y": 2}},
+            "add3": {
+                "process_id": "add3",
+                "namespace": "https://share.test/add3.json",
+                "arguments": {"x": {"from_node": "add12"}},
+                "result": True,
+            },
+        }
+    )
+    assert extract_default_job_options_from_process_graph(pg, processing_mode="batch_job") == {
+        "memory": "2GB",
+        "cpu": "yellow",
+    }
+    assert extract_default_job_options_from_process_graph(pg, processing_mode="synchronous") == {
+        "cpu": "green",
+    }