Make it possible to resolve rx expressions recursively (#918)

param/parameterized.py

@@ -164,11 +164,13 @@ def eval_function_with_deps(function):
             kwargs = {n: getattr(dep.owner, dep.name) for n, dep in kw_deps.items()}
     return function(*args, **kwargs)
 
-def resolve_value(value):
+def resolve_value(value, recursive=True):
     """
     Resolves the current value of a dynamic reference.
     """
-    if isinstance(value, (list, tuple)):
+    if not recursive:
+        pass
+    elif isinstance(value, (list, tuple)):
         return type(value)(resolve_value(v) for v in value)
     elif isinstance(value, dict):
         return type(value)((resolve_value(k), resolve_value(v)) for k, v in value.items())
@@ -2007,14 +2009,15 @@ def _sync_refs(self_, *events):
         updates = {}
         for pname, ref in self_.self._param__private.refs.items():
             # Skip updating value if dependency has not changed
-            deps = resolve_ref(ref, self_[pname].nested_refs)
+            recursive = self_[pname].nested_refs
+            deps = resolve_ref(ref, recursive)
             is_gen = inspect.isgeneratorfunction(ref)
             is_async = iscoroutinefunction(ref) or is_gen
             if not any((dep.owner is e.obj and dep.name == e.name) for dep in deps for e in events) and not is_async:
                 continue
 
             try:
-                new_val = resolve_value(ref)
+                new_val = resolve_value(ref, recursive)
             except Skip:
                 new_val = Undefined
             if new_val is Skip or new_val is Undefined:
@@ -2037,7 +2040,7 @@ def _resolve_ref(self_, pobj, value):
             return None, None, value, False
         ref = value
         try:
-            value = resolve_value(value)
+            value = resolve_value(value, recursive=pobj.nested_refs)
         except Skip:
             value = Undefined
         if is_async:

param/reactive.py

@@ -90,40 +90,92 @@
 from types import FunctionType, MethodType
 from typing import Any, Callable, Optional
 
-from . import Event
 from .depends import depends
 from .display import _display_accessors, _reactive_display_objs
 from .parameterized import (
     Parameter, Parameterized, Skip, Undefined, eval_function_with_deps, get_method_owner,
     register_reference_transform, resolve_ref, resolve_value, transform_reference
 )
+from .parameters import Boolean, Event
 from ._utils import iscoroutinefunction, full_groupby
 
 
 class Wrapper(Parameterized):
     """
-    Simple wrapper to allow updating literal values easily.
+    Helper class to allow updating literal values easily.
     """
 
     object = Parameter(allow_refs=False)
 
 
 class GenWrapper(Parameterized):
     """
-    Wrapper to allow streaming from generator functions.
+    Helper class to allow streaming from generator functions.
     """
 
     object = Parameter(allow_refs=True)
 
 
 class Trigger(Parameterized):
+    """
+    Helper class to allow triggering an event under some condition.
+    """
 
     value = Event()
 
     def __init__(self, parameters, **params):
         super().__init__(**params)
         self.parameters = parameters
 
+class Resolver(Parameterized):
+    """
+    Helper class to allow (recursively) resolving references.
+    """
+
+    object = Parameter(allow_refs=True)
+
+    recursive = Boolean(default=False)
+
+    value = Parameter()
+
+    def __init__(self, **params):
+        self._watchers = []
+        super().__init__(**params)
+
+    def _resolve_value(self, *events):
+        nested = self.param.object.nested_refs
+        refs = resolve_ref(self.object, nested)
+        value = resolve_value(self.object, nested)
+        if self.recursive:
+            new_refs = [r for r in resolve_ref(value, nested) if r not in refs]
+            while new_refs:
+                refs += new_refs
+                value = resolve_value(value, nested)
+                new_refs = [r for r in resolve_ref(value, nested) if r not in refs]
+            if events:
+                self._update_refs(refs)
+        self.value = value
+        return refs
+
+    @depends('object', watch=True, on_init=True)
+    def _resolve_object(self):
+        refs = self._resolve_value()
+        self._update_refs(refs)
+
+    def _update_refs(self, refs):
+        for w in self._watchers:
+            (w.inst or w.cls).param.unwatch(w)
+        self._watchers = []
+        for _, params in full_groupby(refs, lambda x: id(x.owner)):
+            self._watchers.append(
+                params[0].owner.param.watch(self._resolve_value, [p.name for p in params])
+            )
+
+
+class NestedResolver(Resolver):
+
+    object = Parameter(allow_refs=True, nested_refs=True)
+
 
 class reactive_ops:
     """
@@ -235,6 +287,27 @@ def pipe(self, func, /, *args, **kwargs):
         """
         return self._as_rx()._apply_operator(func, *args, **kwargs)
 
+    def resolve(self, nested=True, recursive=False):
+        """
+        Resolves references held by the expression.
+
+        As an example if the expression returns a list of parameters
+        this operation will return a list of the parameter values.
+
+        Arguments
+        ---------
+        nested: bool
+          Whether to resolve references contained within nested objects,
+          i.e. tuples, lists, sets and dictionaries.
+        recursive: bool
+          Whether to recursively resolve references, i.e. if a reference
+          itself returns a reference we recurse into it until no more
+          references can be resolved.
+        """
+        resolver_type = NestedResolver if nested else Resolver
+        resolver = resolver_type(object=self._reactive, recursive=recursive)
+        return resolver.param.value.rx()
+
     def updating(self):
         """
         Returns a new expression that is True while the expression is updating.

tests/testreactive.py

@@ -67,6 +67,8 @@ class Parameters(param.Parameterized):
 
     boolean = param.Boolean(default=False)
 
+    parameter = param.Parameter(allow_refs=False)
+
     event = param.Event()
 
     @param.depends('integer')
@@ -433,6 +435,96 @@ def test_reactive_when_initial():
     p.param.trigger('event')
     assert integer.rx.value == 4
 
+def test_reactive_resolve():
+    p = Parameters(integer=3)
+    p2 = Parameters(parameter=p.param.integer)
+
+    prx = p2.param.parameter.rx()
+    assert prx.rx.value is p.param.integer
+
+    resolved_prx = prx.rx.resolve()
+    assert resolved_prx.rx.value == 3
+
+    changes = []
+    resolved_prx.rx.watch(changes.append)
+
+    # Test changing referenced value
+    p.integer = 4
+    assert resolved_prx.rx.value == 4
+    assert changes == [4]
+
+    # Test changing reference itself
+    p2.parameter = p.param.number
+    assert resolved_prx.rx.value == 3.14
+    assert changes == [4, 3.14]
+
+    # Ensure no updates triggered when old reference is updated
+    p.integer = 5
+    assert resolved_prx.rx.value == 3.14
+    assert changes == [4, 3.14]
+
+def test_reactive_resolve_nested():
+    p = Parameters(integer=3)
+    p2 = Parameters(parameter=[p.param.integer])
+
+    prx = p2.param.parameter.rx()
+    assert prx.rx.value == [p.param.integer]
+
+    resolved_prx = prx.rx.resolve(nested=True)
+    assert resolved_prx.rx.value == [3]
+
+    changes = []
+    resolved_prx.rx.watch(changes.append)
+
+    # Test changing referenced value
+    p.integer = 4
+    assert resolved_prx.rx.value == [4]
+    assert changes == [[4]]
+
+    # Test changing reference itself
+    p2.parameter = [p.param.number]
+    assert resolved_prx.rx.value == [3.14]
+    assert changes == [[4], [3.14]]
+
+    # Ensure no updates triggered when old reference is updated
+    p.integer = 5
+    assert resolved_prx.rx.value == [3.14]
+    assert changes == [[4], [3.14]]
+
+def test_reactive_resolve_recursive():
+    p = Parameters(integer=3)
+    p2 = Parameters(parameter=p.param.integer)
+    p3 = Parameters(parameter=p2.param.parameter)
+
+    prx = p3.param.parameter.rx()
+    assert prx.rx.value is p2.param.parameter
+
+    resolved_prx = prx.rx.resolve(recursive=True)
+    assert resolved_prx.rx.value == 3
+
+    changes = []
+    resolved_prx.rx.watch(changes.append)
+
+    # Test changing referenced value
+    p.integer = 4
+    assert resolved_prx.rx.value == 4
+    assert changes == [4]
+
+    # Test changing recursive reference
+    p2.parameter = p.param.number
+    assert resolved_prx.rx.value == 3.14
+    assert changes == [4, 3.14]
+
+    # Ensure no updates triggered when old reference is updated
+    p.integer = 5
+    assert resolved_prx.rx.value == 3.14
+    assert changes == [4, 3.14]
+
+    # Test changing reference itself
+    p3.parameter = p.param.string
+    assert resolved_prx.rx.value == 'string'
+    assert changes == [4, 3.14, 'string']
+
 async def test_reactive_async_func():
     async def async_func():
         await asyncio.sleep(0.02)