Support passing pg abstractions deeper than 1st argument level

CHANGELOG.md

@@ -12,12 +12,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Add `is_one_of()` to `property_filter` for compact building of an allow-list based property filter.
 
 ### Changed
+
 - `MultiBackendJobManager`: "queued_for_start" is now also considered for the queue limit of a backend.
 
 ### Removed
 
 ### Fixed
+
 - `MultiBackendJobManager`: status "queued_for_start" is no longer overwritten to "created" to allow consistent tracking of the job lifecycle and more accurate handling of job starting.
+- Support passing process graph abstractions (like `DataCube`) deeper than first argument level ([#868](https://github.com/Open-EO/openeo-python-client/issues/868))
 
 
 ## [0.48.0] - 2026-02-17

openeo/internal/graph_building.py

@@ -134,20 +134,27 @@ def __init__(self, process_id: str, arguments: dict = None, namespace: Union[str
         # Merge arguments dict and kwargs
         arguments = dict(**(arguments or {}), **kwargs)
         # Make sure direct PGNode arguments are properly wrapped in a "from_node" dict
-        for arg, value in arguments.items():
-            if isinstance(value, _FromNodeMixin):
-                arguments[arg] = {"from_node": value.from_node()}
-            elif isinstance(value, list):
-                for index, arrayelement in enumerate(value):
-                    if isinstance(arrayelement, _FromNodeMixin):
-                        value[index] = {"from_node": arrayelement.from_node()}
+        arguments = self._wrap_from_nodes(arguments)
         # TODO: use a frozendict of some sort to ensure immutability?
         self._arguments = arguments
         self._namespace = namespace
 
     def from_node(self):
         return self
 
+    def _wrap_from_nodes(self, value):
+        if isinstance(value, _FromNodeMixin):
+            return {"from_node": value.from_node()}
+        elif isinstance(value, list):
+            return [self._wrap_from_nodes(v) for v in value]
+        elif isinstance(value, dict):
+            if "process_graph" in value or "from_node" in value:
+                # Don't recurse into sub-process-graphs or existing "from_node" wrappers
+                return value
+            else:
+                return {k: self._wrap_from_nodes(v) for k, v in value.items()}
+        return value
+
     def __repr__(self):
         return "<{c} {p!r} at 0x{m:x}>".format(c=self.__class__.__name__, p=self.process_id, m=id(self))
 
@@ -422,6 +429,11 @@ def leaveArgument(self, argument_id: str, value):
     def constantArgument(self, argument_id: str, value):
         self._store_argument(argument_id, value)
 
+    def _accept_dict(self, value: dict):
+        for k, v in value.items():
+            if isinstance(v, dict) and "from_node" in v:
+                self.accept_node(v["from_node"])
+
 
 class PGNodeGraphUnflattener(ProcessGraphUnflattener):
     """

tests/rest/datacube/test_datacube100.py

@@ -2984,6 +2984,43 @@ def test_datacube_from_process_no_warnings(con100, caplog, recwarn):
     assert recwarn.list == []
 
 
+def test_datacube_from_process_nesting_direct(con100):
+    """https://github.com/Open-EO/openeo-python-client/issues/868"""
+    settings = con100.datacube_from_process("produce_settings", size=4)
+    cube = con100.datacube_from_process("colorize", settings=settings)
+    assert get_download_graph(cube, drop_save_result=True) == {
+        "producesettings1": {
+            "process_id": "produce_settings",
+            "arguments": {"size": 4},
+        },
+        "colorize1": {
+            "process_id": "colorize",
+            "arguments": {"settings": {"from_node": "producesettings1"}},
+        },
+    }
+
+
+def test_datacube_from_process_nesting_in_dict(con100):
+    """https://github.com/Open-EO/openeo-python-client/issues/868"""
+    settings = con100.datacube_from_process("produce_settings", size=4)
+    cube = con100.datacube_from_process("colorize", context={"color": "red", "settings": settings})
+    assert get_download_graph(cube, drop_save_result=True) == {
+        "producesettings1": {
+            "process_id": "produce_settings",
+            "arguments": {"size": 4},
+        },
+        "colorize1": {
+            "process_id": "colorize",
+            "arguments": {
+                "context": {
+                    "color": "red",
+                    "settings": {"from_node": "producesettings1"},
+                }
+            },
+        },
+    }
+
+
 class TestDataCubeFromFlatGraph:
 
     def test_datacube_from_flat_graph_minimal(self, con100):