Skip to content

Add .rx.awaiting to report whether an expression is still resolving #1176

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
philippjfr wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add .rx.awaiting to report whether an expression is still resolving #1176

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 15 additions & 21 deletions doc/user_guide/Reactive_Expressions.ipynb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,27 +321,7 @@
"cell_type": "markdown",
"id": "227a230b-98b2-4097-8a6d-798ddb63b74a",
"metadata": {},
"source": [
"## Special Methods on `.rx`\n",
"\n",
"To circumvent the limitations explained above, the `.rx` namespace provides reactive versions of the operations that can't be made reactive through overloading:\n",
"\n",
"- `.rx.and_`: Reactive version of `and`.\n",
"- `.rx.bool`: Reactive version of `bool()`.\n",
"- `.rx.in_`: Reactive version of `in`, testing if the value is in the provided collection.\n",
"- `.rx.is_`: Reactive version of `is`, testing the object identity against another object.\n",
"- `.rx.is_not`: Reactive version of `is not`, testing the absence of object identity with another object.\n",
"- `.rx.len`: Reactive version of `len()`, returning the length of the expression\n",
"- `.rx.map`: Applies a function to each item in a collection.\n",
"- `.rx.not_`: Reactive version of `not`.\n",
"- `.rx.or_`: Reactive version of `or`.\n",
"- `.rx.pipe`: Applies the given function (with static or reactive arguments) to this object.\n",
"- `.rx.updating`: Returns a boolean indicating whether the expression is currently updating.\n",
"- `.rx.when`: Generates a new expression that only updates when the provided dependency updates.\n",
"- `.rx.where`: Returns either the first or the second argument, depending on the current value of the expression.\n",
"\n",
"Unlike their corresponding standard Python equivalent, each of these returns a reactive expression that can thus be combined with other reactive expressions to make reactive pipelines."
]
"source": "## Special Methods on `.rx`\n\nTo circumvent the limitations explained above, the `.rx` namespace provides reactive versions of the operations that can't be made reactive through overloading:\n\n- `.rx.and_`: Reactive version of `and`.\n- `.rx.bool`: Reactive version of `bool()`.\n- `.rx.in_`: Reactive version of `in`, testing if the value is in the provided collection.\n- `.rx.is_`: Reactive version of `is`, testing the object identity against another object.\n- `.rx.is_not`: Reactive version of `is not`, testing the absence of object identity with another object.\n- `.rx.len`: Reactive version of `len()`, returning the length of the expression\n- `.rx.map`: Applies a function to each item in a collection.\n- `.rx.not_`: Reactive version of `not`.\n- `.rx.or_`: Reactive version of `or`.\n- `.rx.pipe`: Applies the given function (with static or reactive arguments) to this object.\n- `.rx.updating`: Returns a boolean indicating whether the expression is currently updating.\n- `.rx.when`: Generates a new expression that only updates when the provided dependency updates.\n- `.rx.where`: Returns either the first or the second argument, depending on the current value of the expression.\n\nUnlike their corresponding standard Python equivalent, each of these returns a reactive expression that can thus be combined with other reactive expressions to make reactive pipelines.\n\nThe namespace also provides `.rx.awaiting`, which reports whether an asynchronous operation in the expression is still resolving. Unlike the methods above it is a plain boolean rather than a reactive expression."
},
{
"cell_type": "markdown",
Expand Down Expand Up @@ -604,6 +584,20 @@
"expr.rx.value += 1"
]
},
{
"cell_type": "markdown",
"id": "7a2b23f9",
"source": "#### `.rx.awaiting`\n\nWhile `.rx.updating()` tracks a synchronous computation, an asynchronous operation outlives the update that scheduled it. `.rx.awaiting` reports whether such an operation is still resolving, i.e. whether it has been scheduled but has not yet produced a value for the current inputs.\n\nWhile an expression is awaiting, its value is reported as `Undefined` rather than the value it computed from the previous inputs, so `awaiting` is what distinguishes \"not resolved yet\" from an operation that deliberately skipped. The whole graph feeding the expression is considered, so a synchronous operation downstream of an asynchronous one also reports `True` while its input resolves:",
"metadata": {}
},
{
"cell_type": "code",
"id": "1d5991d8",
"source": "import asyncio\n\nasync def slow_double(value):\n await asyncio.sleep(1)\n return value * 2\n\nasync_expr = rx(1).rx.pipe(slow_double) + 1\n\nprint(f'requested: value={async_expr.rx.value!r}, awaiting={async_expr.rx.awaiting}')\n\nawait asyncio.sleep(1.5)\n\nprint(f'settled: value={async_expr.rx.value!r}, awaiting={async_expr.rx.awaiting}')",
"metadata": {},
"execution_count": null,
"outputs": []
},
{
"cell_type": "markdown",
"id": "dfea7f35-f0fa-4bc0-954f-b01d5dcf9d6c",
Expand Down
137 changes: 137 additions & 0 deletions param/reactive.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -763,6 +763,67 @@ def resolve(self, nested=True, recursive=False) -> 'rx':
resolver = resolver_type(object=self._reactive, recursive=recursive)
return resolver.param.value.rx()

@property
def awaiting(self) -> bool:
"""
Whether any asynchronous operation in this expression is still resolving.

``True`` from the moment an asynchronous operation is scheduled until it
produces a value for the current inputs. While a node is awaiting, the
expression still holds the value it computed from the previous inputs,
so ``.rx.value`` reports :obj:`param.Undefined` rather than that stale
value; ``awaiting`` distinguishes "not resolved yet" from an operation
that deliberately skipped.

The whole graph feeding the expression is considered, not just the node
it is accessed on, so a synchronous operation downstream of an
asynchronous one reports ``True`` while its input resolves.

Reading this does not itself schedule anything, so an expression whose
value has never been requested reports ``False`` until something asks
for it.

Both routes an asynchronous callable can take are tracked: one applied
as an operation, e.g. passed to ``.rx.pipe``, and one passed to ``rx``
as the object itself, which is held on a parameter and resolved by the
reference machinery. A generator settles on each value it yields, so it
reports ``True`` only until its next value arrives rather than until it
is exhausted. Accessed on a parameter rather than an expression this is
always ``False``.

Returns
-------
bool
``True`` while an asynchronous operation has not yet produced a
value for the current inputs, ``False`` otherwise.

Examples
--------
Pipe through a coroutine function and observe the expression settle:

>>> import asyncio, param
>>> async def double(value):
... await asyncio.sleep(0.1)
... return value * 2
>>> expr = param.rx(1).rx.pipe(double) + 1

Requesting the value schedules the operation:

>>> expr.rx.value is param.Undefined
True
>>> expr.rx.awaiting
True

Once the coroutine has resolved the expression reports a value again:

>>> expr.rx.awaiting # doctest: +SKIP
False
"""
reactive = self._reactive
if not isinstance(reactive, rx):
return False
return any(node._settling for node in reactive._upstream())

def updating(self) -> 'rx':
"""
Return a new expression that indicates whether the current expression is updating.
Expand Down Expand Up @@ -1707,9 +1768,64 @@ def _awaiting(self) -> bool:
"""
Whether an asynchronous resolution is in flight that has not yet
produced a value for the current generation.

While a node is awaiting, the cached ``_current_`` value was computed
from inputs that have since been superseded, so resolving the node
skips instead of reporting the stale value as if it were current.
"""
return self._resolve_generation != self._finished_generation

@property
def _awaiting_ref(self) -> bool:
"""
Whether an asynchronous reference feeding this node has not yet
produced a value for the current inputs.

A coroutine or generator function passed to ``rx`` as the object rather
than as an operation is held on a parameter and resolved by the
reference machinery, so its settlement is tracked there instead of by
this node's own generations.
"""
for p in self._internal_params:
owner, name = p.owner, p.name
if name is None or not isinstance(owner, Parameterized):
continue
if owner.param._awaiting_ref(name):
return True
return False

@property
def _settling(self) -> bool:
"""Whether this node is waiting on an asynchronous result of its own."""
return self._awaiting or self._awaiting_ref

def _upstream(self) -> Iterator[rx]:
"""
Yield this node and every ``rx`` node it derives its value from.

Inputs reach a node by three routes, all of which have to be visited
because an operation is only as settled as the nodes feeding it: the
``_prev`` chain of the pipeline the node belongs to, the ``_shared``
input it was cloned from when a pipeline branches, and any ``rx``
passed as an argument to one of its operations.
"""
seen: set[int] = set()
stack: list[rx] = [self]
while stack:
node = stack.pop()
if id(node) in seen:
continue
seen.add(id(node))
yield node
for inp in (node._prev, node._shared):
if isinstance(inp, rx):
stack.append(inp)
operation = node._operation
if operation:
stack.extend(_iter_rx((
operation['fn'], operation.get('args', ()), operation.get('kwargs', {})
)))

@property
def _current(self):
if self._error_state:
Expand Down Expand Up @@ -2277,6 +2393,27 @@ def __setattr__(self, name, value):
super().__setattr__(name, value)


def _iter_rx(value: t.Any) -> Iterator[rx]:
"""
Yield the reactive expressions nested anywhere inside a reference.

Mirrors the containers ``resolve_value`` descends into, so an ``rx`` used
as an operation argument is found wherever ``resolve_value`` would find it.
"""
if isinstance(value, rx):
yield value
elif isinstance(value, (list, tuple, set)):
for v in value:
yield from _iter_rx(v)
elif isinstance(value, dict):
for k, v in value.items():
yield from _iter_rx(k)
yield from _iter_rx(v)
elif isinstance(value, slice):
for v in (value.start, value.stop, value.step):
yield from _iter_rx(v)


def _rx_transform(obj):
if not isinstance(obj, rx):
return obj
Expand Down
Loading
Loading