From f86e60f0f5afab07479d970ad0b2c253ec0362c3 Mon Sep 17 00:00:00 2001 From: Wei8007 <189627153@qq.com> Date: Mon, 13 Jul 2026 20:24:42 +0800 Subject: [PATCH] Add recursive process graph unrolling --- CHANGELOG.md | 1 + openeo/internal/graph_unrolling.py | 220 +++++++++++++++++++++++++ openeo/rest/connection.py | 79 +++++++++ tests/internal/test_graph_unrolling.py | 182 ++++++++++++++++++++ tests/rest/test_connection.py | 91 ++++++++++ 5 files changed, 573 insertions(+) create mode 100644 openeo/internal/graph_unrolling.py create mode 100644 tests/internal/test_graph_unrolling.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e570aba6e..554429b06 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Add `Connection.unroll_process_graph()` to recursively inline user-defined and remote process definitions. ([#749](https://github.com/Open-EO/openeo-python-client/issues/749)) - `DataCube.resample_spatial()` now supports parameterized `resolution` and `projection` arguments. ([#897](https://github.com/Open-EO/openeo-python-client/issues/897)) - Sanitize asset download filenames (e.g. strip slashes, (semi)colon, hash, ...), instead of blindly using the asset key as filename. ([#820](https://github.com/Open-EO/openeo-python-client/issues/820)) - Support parameters in `DataCube` apply- and band-math operations ([#903](https://github.com/Open-EO/openeo-python-client/issues/903)) diff --git a/openeo/internal/graph_unrolling.py b/openeo/internal/graph_unrolling.py new file mode 100644 index 000000000..49da38dec --- /dev/null +++ b/openeo/internal/graph_unrolling.py @@ -0,0 +1,220 @@ +from __future__ import annotations + +import copy +from typing import Callable, Dict, Iterable, Mapping, Optional, Tuple + + +class ProcessGraphUnrollError(ValueError): + """Raised when a process graph can not be unrolled safely.""" + + +ProcessDefinitionResolver = Callable[[Mapping], Optional[Mapping]] + + +class ProcessGraphUnroller: + """Inline user-defined process definitions into a flat process graph.""" + + def __init__(self, *, resolve_process_definition: ProcessDefinitionResolver): + self._resolve_process_definition = resolve_process_definition + + def unroll(self, process_graph: Mapping[str, Mapping]) -> Dict[str, dict]: + return self._unroll_graph(copy.deepcopy(dict(process_graph)), stack=()) + + def _unroll_graph( + self, process_graph: Dict[str, dict], *, stack: Tuple[Tuple[str, Optional[str]], ...] + ) -> Dict[str, dict]: + process_graph = self._unroll_callback_graphs(process_graph=process_graph, stack=stack) + + while True: + expandable = next( + ( + (node_id, node, definition) + for node_id, node in process_graph.items() + if (definition := self._resolve_process_definition(node)) is not None + ), + None, + ) + if expandable is None: + return process_graph + + node_id, node, definition = expandable + process_key = (str(node.get("process_id")), node.get("namespace")) + if process_key in stack: + chain = " -> ".join(pid for pid, _ in (*stack, process_key)) + raise ProcessGraphUnrollError(f"Recursive process definition detected: {chain}") + + inlined = self._instantiate_definition( + definition=definition, + arguments=node.get("arguments", {}), + invocation_id=node_id, + used_node_ids=process_graph, + ) + inlined = self._unroll_graph(inlined, stack=(*stack, process_key)) + result_id = self._get_result_node_id(inlined) + + if node.get("result"): + inlined[result_id]["result"] = True + else: + inlined[result_id].pop("result", None) + + process_graph = self._replace_node( + process_graph=process_graph, + node_id=node_id, + replacement=inlined, + replacement_result_id=result_id, + ) + process_graph = self._unroll_callback_graphs(process_graph=process_graph, stack=stack) + + def _instantiate_definition( + self, + *, + definition: Mapping, + arguments: Mapping, + invocation_id: str, + used_node_ids: Iterable[str], + ) -> Dict[str, dict]: + process_graph = definition.get("process_graph") + if not isinstance(process_graph, Mapping) or not process_graph: + raise ProcessGraphUnrollError( + f"Process definition {definition.get('id', invocation_id)!r} has no process graph" + ) + + bindings = self._build_parameter_bindings(definition=definition, arguments=arguments) + node_id_map = self._build_node_id_map( + inner_node_ids=process_graph, + invocation_id=invocation_id, + used_node_ids=used_node_ids, + ) + + instantiated = {} + for inner_id, inner_node in copy.deepcopy(dict(process_graph)).items(): + inner_node = self._rewrite_references(inner_node, node_id_map) + inner_node = self._bind_parameters(inner_node, bindings) + instantiated[node_id_map[inner_id]] = inner_node + return instantiated + + @staticmethod + def _build_parameter_bindings(*, definition: Mapping, arguments: Mapping) -> Dict[str, object]: + parameters = definition.get("parameters", []) + if not isinstance(parameters, list): + raise ProcessGraphUnrollError(f"Invalid parameters in process definition {definition.get('id')!r}") + + bindings = {} + missing = [] + for parameter in parameters: + name = parameter.get("name") + if not isinstance(name, str): + raise ProcessGraphUnrollError(f"Invalid parameter in process definition {definition.get('id')!r}") + if name in arguments: + bindings[name] = copy.deepcopy(arguments[name]) + elif "default" in parameter: + bindings[name] = copy.deepcopy(parameter["default"]) + elif parameter.get("optional", False): + bindings[name] = None + else: + missing.append(name) + + if missing: + raise ProcessGraphUnrollError( + f"Missing required arguments {sorted(missing)} for process {definition.get('id')!r}" + ) + return bindings + + @staticmethod + def _build_node_id_map( + *, inner_node_ids: Iterable[str], invocation_id: str, used_node_ids: Iterable[str] + ) -> Dict[str, str]: + used = set(used_node_ids) + result = {} + for inner_id in inner_node_ids: + base = f"{invocation_id}_{inner_id}" + candidate = base + index = 2 + while candidate in used: + candidate = f"{base}_{index}" + index += 1 + result[inner_id] = candidate + used.add(candidate) + return result + + def _unroll_callback_graphs( + self, *, process_graph: Dict[str, dict], stack: Tuple[Tuple[str, Optional[str]], ...] + ) -> Dict[str, dict]: + def visit(value): + if isinstance(value, dict): + return { + key: ( + self._unroll_graph(copy.deepcopy(child), stack=stack) + if key == "process_graph" and isinstance(child, dict) + else visit(child) + ) + for key, child in value.items() + } + if isinstance(value, list): + return [visit(child) for child in value] + return value + + return {node_id: visit(node) for node_id, node in process_graph.items()} + + @staticmethod + def _rewrite_references(value, node_id_map: Mapping[str, str]): + if isinstance(value, dict): + if set(value) == {"from_node"} and value["from_node"] in node_id_map: + return {"from_node": node_id_map[value["from_node"]]} + return { + key: child if key == "process_graph" else ProcessGraphUnroller._rewrite_references(child, node_id_map) + for key, child in value.items() + } + if isinstance(value, list): + return [ProcessGraphUnroller._rewrite_references(child, node_id_map) for child in value] + return value + + @staticmethod + def _bind_parameters(value, bindings: Mapping[str, object]): + if isinstance(value, dict): + if set(value) == {"from_parameter"}: + parameter = value["from_parameter"] + if parameter not in bindings: + raise ProcessGraphUnrollError(f"Unknown process parameter {parameter!r}") + return copy.deepcopy(bindings[parameter]) + return { + key: child if key == "process_graph" else ProcessGraphUnroller._bind_parameters(child, bindings) + for key, child in value.items() + } + if isinstance(value, list): + return [ProcessGraphUnroller._bind_parameters(child, bindings) for child in value] + return value + + @staticmethod + def _get_result_node_id(process_graph: Mapping[str, Mapping]) -> str: + result_nodes = [node_id for node_id, node in process_graph.items() if node.get("result") is True] + if len(result_nodes) != 1: + raise ProcessGraphUnrollError(f"Expected one result node, found {len(result_nodes)}") + return result_nodes[0] + + @staticmethod + def _replace_node( + *, + process_graph: Dict[str, dict], + node_id: str, + replacement: Dict[str, dict], + replacement_result_id: str, + ) -> Dict[str, dict]: + def replace_reference(value): + if isinstance(value, dict): + if value == {"from_node": node_id}: + return {"from_node": replacement_result_id} + return { + key: child if key == "process_graph" else replace_reference(child) for key, child in value.items() + } + if isinstance(value, list): + return [replace_reference(child) for child in value] + return value + + result = {} + for current_id, current_node in process_graph.items(): + if current_id == node_id: + result.update(replacement) + else: + result[current_id] = replace_reference(current_node) + return result diff --git a/openeo/rest/connection.py b/openeo/rest/connection.py index 62d329f2b..d7cb1bb6d 100644 --- a/openeo/rest/connection.py +++ b/openeo/rest/connection.py @@ -41,6 +41,10 @@ _FromNodeMixin, as_flat_graph, ) +from openeo.internal.graph_unrolling import ( + ProcessGraphUnroller, + ProcessGraphUnrollError, +) from openeo.internal.jupyter import VisualDict, VisualList from openeo.internal.processes.builder import ProcessBuilderBase from openeo.internal.warnings import deprecated, legacy_alias @@ -1024,6 +1028,81 @@ def describe_process(self, id: str, namespace: Optional[str] = None) -> dict: raise OpenEoClientException("Process does not exist.") + def unroll_process_graph( + self, + process_graph: Union[dict, FlatGraphableMixin, str, Path, List[FlatGraphableMixin]], + ) -> Dict[str, dict]: + """ + Inline user-defined and remote process definitions in a process graph. + + The returned flat process graph only references processes advertised by the + backend. Registered user-defined processes are loaded from ``/process_graphs``; + process nodes with an HTTP(S) ``namespace`` are loaded as remote process + definitions. Nested definitions are expanded recursively and parameter + defaults are applied. + + :param process_graph: process graph or graph-building object to unroll. + :return: a new flat process graph. The input is not modified. + + .. versionadded:: 0.51.0 + """ + graph = as_flat_graph(process_graph) + if isinstance(graph.get("process"), dict) and isinstance(graph["process"].get("process_graph"), dict): + graph = graph["process"]["process_graph"] + elif isinstance(graph.get("process_graph"), dict): + graph = graph["process_graph"] + + definition_cache: Dict[Tuple[str, Optional[str]], Optional[dict]] = {} + predefined_process_ids: Optional[Set[str]] = None + + def extract_remote_definition(data: dict, *, process_id: str, namespace: str) -> dict: + if "process_graph" in data: + definition = data + elif isinstance(data.get("processes"), list): + matches = [process for process in data["processes"] if process.get("id") == process_id] + if len(matches) != 1: + raise ProcessGraphUnrollError( + f"Expected one definition for process {process_id!r} at {namespace!r}, found {len(matches)}" + ) + definition = matches[0] + else: + raise ProcessGraphUnrollError(f"No process definition found at {namespace!r}") + + if definition.get("id") not in {None, process_id}: + raise ProcessGraphUnrollError( + f"Expected process {process_id!r} at {namespace!r}, found {definition.get('id')!r}" + ) + return definition + + def resolve(node: Mapping) -> Optional[dict]: + nonlocal predefined_process_ids + + process_id = node.get("process_id") + namespace = node.get("namespace") + if not isinstance(process_id, str): + return None + key = (process_id, namespace) + if key in definition_cache: + return definition_cache[key] + + if isinstance(namespace, str) and urllib.parse.urlparse(namespace).scheme in {"http", "https"}: + response = requests.get(namespace) + response.raise_for_status() + definition = extract_remote_definition(response.json(), process_id=process_id, namespace=namespace) + elif namespace is not None: + definition = None + else: + if predefined_process_ids is None: + predefined_process_ids = {process["id"] for process in self.list_processes()} + definition = ( + None if process_id in predefined_process_ids else self.user_defined_process(process_id).describe() + ) + + definition_cache[key] = definition + return definition + + return ProcessGraphUnroller(resolve_process_definition=resolve).unroll(graph) + def list_jobs(self, limit: Union[int, None] = 100) -> JobListingResponse: """ Lists (batch) jobs metadata of the authenticated user. diff --git a/tests/internal/test_graph_unrolling.py b/tests/internal/test_graph_unrolling.py new file mode 100644 index 000000000..7acb9d6d8 --- /dev/null +++ b/tests/internal/test_graph_unrolling.py @@ -0,0 +1,182 @@ +import pytest + +from openeo.internal.graph_unrolling import ( + ProcessGraphUnroller, + ProcessGraphUnrollError, +) + + +def test_unroll_process_with_parameters_and_default(): + definitions = { + "increment": { + "id": "increment", + "parameters": [ + {"name": "data"}, + {"name": "amount", "default": 1}, + ], + "process_graph": { + "add1": { + "process_id": "add", + "arguments": { + "x": {"from_parameter": "data"}, + "y": {"from_parameter": "amount"}, + }, + "result": True, + } + }, + } + } + unroller = ProcessGraphUnroller(resolve_process_definition=lambda node: definitions.get(node["process_id"])) + graph = { + "load1": {"process_id": "load_collection", "arguments": {"id": "S2"}}, + "increment1": { + "process_id": "increment", + "arguments": {"data": {"from_node": "load1"}}, + }, + "save1": { + "process_id": "save_result", + "arguments": {"data": {"from_node": "increment1"}, "format": "GTiff"}, + "result": True, + }, + } + + result = unroller.unroll(graph) + + assert result == { + "load1": {"process_id": "load_collection", "arguments": {"id": "S2"}}, + "increment1_add1": { + "process_id": "add", + "arguments": {"x": {"from_node": "load1"}, "y": 1}, + }, + "save1": { + "process_id": "save_result", + "arguments": {"data": {"from_node": "increment1_add1"}, "format": "GTiff"}, + "result": True, + }, + } + assert graph["increment1"]["process_id"] == "increment" + + +def test_unroll_nested_processes(): + definitions = { + "increment": { + "id": "increment", + "parameters": [{"name": "data"}], + "process_graph": { + "add1": { + "process_id": "add", + "arguments": {"x": {"from_parameter": "data"}, "y": 1}, + "result": True, + } + }, + }, + "increment_twice": { + "id": "increment_twice", + "parameters": [{"name": "data"}], + "process_graph": { + "first": { + "process_id": "increment", + "arguments": {"data": {"from_parameter": "data"}}, + }, + "second": { + "process_id": "increment", + "arguments": {"data": {"from_node": "first"}}, + "result": True, + }, + }, + }, + } + unroller = ProcessGraphUnroller(resolve_process_definition=lambda node: definitions.get(node["process_id"])) + + result = unroller.unroll( + { + "twice1": { + "process_id": "increment_twice", + "arguments": {"data": 5}, + "result": True, + } + } + ) + + assert result == { + "twice1_first_add1": {"process_id": "add", "arguments": {"x": 5, "y": 1}}, + "twice1_second_add1": { + "process_id": "add", + "arguments": {"x": {"from_node": "twice1_first_add1"}, "y": 1}, + "result": True, + }, + } + + +def test_unroll_recursive_process_fails(): + definition = { + "id": "recursive", + "parameters": [], + "process_graph": {"again": {"process_id": "recursive", "arguments": {}, "result": True}}, + } + unroller = ProcessGraphUnroller( + resolve_process_definition=lambda node: definition if node["process_id"] == "recursive" else None + ) + + with pytest.raises(ProcessGraphUnrollError, match="Recursive process definition detected: recursive -> recursive"): + unroller.unroll({"recursive1": {"process_id": "recursive", "arguments": {}, "result": True}}) + + +def test_unroll_process_in_callback_scope(): + increment = { + "id": "increment", + "parameters": [{"name": "data"}], + "process_graph": { + "add1": { + "process_id": "add", + "arguments": {"x": {"from_parameter": "data"}, "y": 1}, + "result": True, + } + }, + } + unroller = ProcessGraphUnroller( + resolve_process_definition=lambda node: increment if node["process_id"] == "increment" else None + ) + + result = unroller.unroll( + { + "apply1": { + "process_id": "apply", + "arguments": { + "data": {"from_node": "load1"}, + "process": { + "process_graph": { + "increment1": { + "process_id": "increment", + "arguments": {"data": {"from_parameter": "x"}}, + "result": True, + } + } + }, + }, + "result": True, + }, + "load1": {"process_id": "load_collection", "arguments": {"id": "S2"}}, + } + ) + + callback = result["apply1"]["arguments"]["process"]["process_graph"] + assert callback == { + "increment1_add1": { + "process_id": "add", + "arguments": {"x": {"from_parameter": "x"}, "y": 1}, + "result": True, + } + } + + +def test_unroll_missing_required_argument_fails(): + definition = { + "id": "increment", + "parameters": [{"name": "data"}], + "process_graph": {"add1": {"process_id": "add", "arguments": {}, "result": True}}, + } + unroller = ProcessGraphUnroller(resolve_process_definition=lambda node: definition) + + with pytest.raises(ProcessGraphUnrollError, match=r"Missing required arguments \['data'\]"): + unroller.unroll({"increment1": {"process_id": "increment", "arguments": {}, "result": True}}) diff --git a/tests/rest/test_connection.py b/tests/rest/test_connection.py index c5d425336..2b33f6f77 100644 --- a/tests/rest/test_connection.py +++ b/tests/rest/test_connection.py @@ -4167,6 +4167,97 @@ def test_describe_processes(requests_mock): conn.describe_process("invalid") +def test_unroll_process_graph_registered_udp(requests_mock): + requests_mock.get(API_URL, json=build_capabilities(api_version="1.0.0", udp=True)) + processes = requests_mock.get(API_URL + "processes", json={"processes": [{"id": "add"}]}) + udp = requests_mock.get( + API_URL + "process_graphs/increment", + json={ + "id": "increment", + "parameters": [{"name": "data"}, {"name": "amount", "default": 1}], + "process_graph": { + "add1": { + "process_id": "add", + "arguments": { + "x": {"from_parameter": "data"}, + "y": {"from_parameter": "amount"}, + }, + "result": True, + } + }, + }, + ) + conn = Connection(API_URL) + + result = conn.unroll_process_graph( + { + "process_graph": { + "increment1": {"process_id": "increment", "arguments": {"data": 3}}, + "increment2": { + "process_id": "increment", + "arguments": {"data": {"from_node": "increment1"}, "amount": 5}, + "result": True, + }, + } + } + ) + + assert result == { + "increment1_add1": {"process_id": "add", "arguments": {"x": 3, "y": 1}}, + "increment2_add1": { + "process_id": "add", + "arguments": {"x": {"from_node": "increment1_add1"}, "y": 5}, + "result": True, + }, + } + assert processes.call_count == 1 + assert udp.call_count == 1 + + +def test_unroll_process_graph_remote_definition(requests_mock): + requests_mock.get(API_URL, json=build_capabilities(api_version="1.0.0")) + requests_mock.get(API_URL + "processes", json={"processes": [{"id": "multiply"}]}) + remote = requests_mock.get( + "https://processes.test/definitions", + json={ + "processes": [ + { + "id": "double", + "parameters": [{"name": "x"}], + "process_graph": { + "multiply1": { + "process_id": "multiply", + "arguments": {"x": {"from_parameter": "x"}, "y": 2}, + "result": True, + } + }, + } + ] + }, + ) + conn = Connection(API_URL) + + result = conn.unroll_process_graph( + { + "double1": { + "process_id": "double", + "namespace": "https://processes.test/definitions", + "arguments": {"x": 21}, + "result": True, + } + } + ) + + assert result == { + "double1_multiply1": { + "process_id": "multiply", + "arguments": {"x": 21, "y": 2}, + "result": True, + } + } + assert remote.call_count == 1 + + def test_list_processes_namespace(requests_mock): requests_mock.get(API_URL, json={"api_version": "1.0.0"}) processes = [{"id": "add"}, {"id": "mask"}]