diff --git a/doc/user_guide/Generators.ipynb b/doc/user_guide/Generators.ipynb index 878643420..9664d6878 100644 --- a/doc/user_guide/Generators.ipynb +++ b/doc/user_guide/Generators.ipynb @@ -140,7 +140,7 @@ "id": "63e76938-6eac-4f28-8362-7b9637ed935f", "metadata": {}, "source": [ - "Note that `rx` can also evaluate a generator directly:" + "Note that `rx` can also evaluate a generator. To make it a reference we wrap it with `param.bind`, since passing the bare generator function to `rx` is deprecated and will eventually wrap the function itself, just like any other function:" ] }, { @@ -150,7 +150,7 @@ "metadata": {}, "outputs": [], "source": [ - "gauge_rx = param.rx(poll_sensor)\n", + "gauge_rx = param.rx(param.bind(poll_sensor))\n", "\n", "gauge_rx" ] diff --git a/doc/user_guide/Reactive_Expressions.ipynb b/doc/user_guide/Reactive_Expressions.ipynb index 3f64e8a6e..027a15061 100644 --- a/doc/user_guide/Reactive_Expressions.ipynb +++ b/doc/user_guide/Reactive_Expressions.ipynb @@ -928,6 +928,37 @@ "execution_count": null, "outputs": [] }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Calling an expression that holds an asynchronous function, or an asynchronous method of the object it wraps, awaits the result in the same way:" + ], + "id": "call-async-md" + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "class Doubler:\n", + " async def double(self, value):\n", + " await asyncio.sleep(1)\n", + " return value * 2\n", + "\n", + "number = rx(1)\n", + "called = rx(slow_double)(number)\n", + "method = rx(Doubler()).double(number)\n", + "\n", + "print(f'requested: called={called.rx.value!r}, method={method.rx.value!r}')\n", + "\n", + "await asyncio.sleep(1.5)\n", + "\n", + "print(f'settled: called={called.rx.value!r}, method={method.rx.value!r}')" + ], + "id": "call-async-code" + }, { "cell_type": "markdown", "id": "9c4f1a52", diff --git a/param/parameterized.py b/param/parameterized.py index d577ec50f..b3bc6aa18 100644 --- a/param/parameterized.py +++ b/param/parameterized.py @@ -335,8 +335,8 @@ def resolve_value(value: t.Any, recursive: bool = True) -> t.Any: resolve_value(value.step) ) value = transform_reference(value) - is_gen = inspect.isgeneratorfunction(value) - if hasattr(value, '_dinfo') or iscoroutinefunction(value) or is_gen: + if hasattr(value, '_dinfo') or _is_async_fn(value): + is_gen = inspect.isgeneratorfunction(value) value = eval_function_with_deps(value) if is_gen: value = _to_async_gen(value) @@ -344,6 +344,51 @@ def resolve_value(value: t.Any, recursive: bool = True) -> t.Any: value = getattr(value.owner, value.name) return value +def _is_async_fn(value: t.Any) -> bool: + """Whether value is a coroutine or generator function.""" + return iscoroutinefunction(value) or inspect.isgeneratorfunction(value) + +def _is_bare_async_ref(value: t.Any) -> bool: + """ + Whether value is a coroutine or generator function that has not been + declared a reference with ``bind`` or ``depends``. + """ + return not hasattr(value, '_dinfo') and _is_async_fn(value) + +def _warn_bare_async_ref(value: t.Any, operand: bool = False) -> None: + """ + Warn that rx will stop calling the bare coroutine or generator function + ``value``. ``operand`` marks an input to an rx operation, which is called + but whose result is never awaited or iterated. + """ + if inspect.isasyncgenfunction(value): + kind = 'async generator' + elif inspect.isgeneratorfunction(value): + kind = 'generator' + else: + kind = 'coroutine' + func = value + while isinstance(func, partial): + func = func.func + name = getattr(func, '__name__', repr(func)) + if operand: + result = 'coroutine' if kind == 'coroutine' else 'async generator' + message = ( + f"Passing the {kind} function {name!r} as an input to a reactive " + f"operation is deprecated. It is currently called and the resulting " + f"{result} is passed on without being awaited or iterated; in a " + "future version the function itself will be passed on." + ) + else: + bound = f"param.bind({name}, ...)" if func is not value else f"param.bind({name})" + message = ( + f"Passing the {kind} function {name!r} to a reactive expression is " + "deprecated. In a future version rx will treat it like any other " + "function, i.e. keep it as a value instead of calling and resolving " + f"it. Wrap it with {bound} to keep resolving it." + ) + warnings.warn(message, category=_ParamFutureWarning, stacklevel=_find_stack_level()) + def resolve_ref(reference: t.Any, recursive: bool = False) -> list[Parameter]: """Resolve all parameters a dynamic reference depends on.""" if recursive: @@ -2975,7 +3020,7 @@ def _setup_params(self_, **params): # speculatively evaluate values to check whether they # contain a reference and warn the user that the # behavior may change in future. - if name not in self_.cls._param__private.explicit_no_refs: + if name not in self_.cls._param__private.explicit_no_refs and not _is_bare_async_ref(val): resolved = val try: ref, _, resolved, _, _ = self_._resolve_ref(pobj, val) @@ -3050,8 +3095,7 @@ def _sync_refs(self_, *events): # Skip updating value if dependency has not changed recursive = self_[pname].nested_refs deps = resolve_ref(ref, recursive) - is_gen = inspect.isgeneratorfunction(ref) - is_async = iscoroutinefunction(ref) or is_gen + is_async = _is_async_fn(ref) 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 @@ -3080,10 +3124,9 @@ def _sync_refs(self_, *events): _notify_async_ref_settle_change(self_.self, pname) def _resolve_ref(self_, pobj: Parameter, value: t.Any): - is_gen = inspect.isgeneratorfunction(value) - is_async = iscoroutinefunction(value) or is_gen + is_async = _is_async_fn(value) deps = resolve_ref(value, recursive=pobj.nested_refs) - if not (deps or is_async or is_gen): + if not (deps or is_async): return None, None, value, False, False ref = value try: @@ -3140,8 +3183,7 @@ def _awaiting_ref(self_, pname: str) -> bool: return False private = self_.self._param__private ref = private.refs.get(pname) - is_async = iscoroutinefunction(ref) or inspect.isgeneratorfunction(ref) - return is_async and private.async_ref_scheduled[pname] != private.async_ref_settled[pname] + return _is_async_fn(ref) and private.async_ref_scheduled[pname] != private.async_ref_settled[pname] async def _async_ref(self_, pname: str, awaitable: t.Awaitable[t.Any], generation: int = 0): if self_.self is None: diff --git a/param/reactive.py b/param/reactive.py index 77a496e97..6f7b45b73 100644 --- a/param/reactive.py +++ b/param/reactive.py @@ -116,8 +116,9 @@ from .parameterized import ( Comparator, Parameter, Parameterized, Skip, Undefined, eval_function_with_deps, get_method_owner, register_reference_transform, resolve_ref, resolve_value, - transform_reference, _notify_weak_listeners, _register_weak_listener, - _watch_async_ref_settle_change, _watch_ref_change, _WeakInvalidator + transform_reference, _is_async_fn, _is_bare_async_ref, _notify_weak_listeners, + _register_weak_listener, _warn_bare_async_ref, _watch_async_ref_settle_change, + _watch_ref_change, _WeakInvalidator ) from .parameters import Boolean, Event, String from ._utils import _thread_state, _to_async_gen, iscoroutinefunction, full_groupby @@ -177,6 +178,22 @@ class GenWrapper(Parameterized): object: t.Any = Parameter(allow_refs=True) +class _HeldValue: + """ + Box for a function assigned to a ``GenWrapper`` root that rx keeps as a + value, so that the reference machinery does not call it. + """ + + __slots__ = ('value',) + + def __init__(self, value: t.Any): + self.value = value + + +def _unbox_held(obj: t.Any) -> t.Any: + return obj.value if isinstance(obj, _HeldValue) else obj + + class Trigger(Parameterized): """Helper class to allow triggering an event under some condition.""" @@ -427,6 +444,7 @@ def __getitem__(self, key: t.Any) -> t.Any: def _set(self, key: t.Any, value: t.Any): """Mask a resolved ``key`` without invalidating the node.""" + _warn_bare_async_inputs(value) node = self._node operation = t.cast('dict', node._operation) overrides = operation.get('overrides') @@ -1400,7 +1418,8 @@ def awaiting(self) -> builtins.bool: 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 + as the object itself after declaring it a reference with + ``param.bind``, 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 @@ -1860,6 +1879,7 @@ def where(self, x, y) -> 'rx': >>> rx_result.rx.value '5 or below' """ + _warn_bare_async_inputs(x, y) xrefs = resolve_ref(x) yrefs = resolve_ref(y) if isinstance(self._reactive, rx): @@ -1879,7 +1899,7 @@ def trigger_y(*args): bind(trigger_y, *yrefs, watch=True) def ternary(condition, _): - return resolve_value(x) if condition else resolve_value(y) + return _resolve_operand(x) if condition else _resolve_operand(y) return rx(bind(ternary, self._reactive, trigger.param.value)) def distinct(self, fn=None) -> 'rx': @@ -2064,7 +2084,12 @@ def value(self, new): "expression wrapping a Parameter or another dynamic " "reference cannot be updated." ) - self._reactive._wrapper.object = resolve_value(new) + _warn_bare_async_inputs(new) + wrapper = self._reactive._wrapper + value = _resolve_operand(new) + if isinstance(wrapper, GenWrapper) and _held_as_value(transform_reference(value)): + value = _HeldValue(value) + wrapper.object = value def watch(self, fn=None, onlychanged=True, queued=False, precedence=0, process_failures=False): """ @@ -2590,6 +2615,7 @@ def collect(*args, error_mode='raise', **kwargs) -> 'rx': >>> collected.args.rx.pipe(sum).rx.value 3 """ + _warn_bare_async_inputs(*args, *kwargs.values()) operation = { 'fn': _collect_marker, 'args': args, @@ -2653,8 +2679,10 @@ def _untracked_reference(value) -> bool: return False if isinstance(value, (list, tuple, dict, slice)): return any(_iter_rx(value)) or bool(resolve_ref(value, recursive=True)) - if hasattr(value, '_dinfo') or iscoroutinefunction(value) or inspect.isgeneratorfunction(value): + if hasattr(value, '_dinfo'): return True + if _is_async_fn(value): + return not _held_as_value(value) return isinstance(value, Parameter) and value.name is not None @@ -2844,10 +2872,12 @@ def __new__(cls, obj=None, **kwargs): # rx._clone codepath fn = kwargs.pop('fn') wrapper = kwargs.pop('_wrapper', None) - elif inspect.isgeneratorfunction(obj) or iscoroutinefunction(obj): + elif _is_async_fn(obj) and not _held_as_value(obj): # Resolves generator and coroutine functions lazily + if _is_bare_async_ref(obj): + _warn_bare_async_ref(obj) wrapper = t.cast('Callable', GenWrapper)(object=obj) - fn = bind(lambda obj: obj, wrapper.param.object) + fn = bind(_unbox_held, wrapper.param.object) obj = Undefined elif isinstance(obj, (FunctionType, MethodType)) and hasattr(obj, '_dinfo'): # Bound functions and methods are resolved on access @@ -2923,7 +2953,7 @@ def __init__( # Define special trigger parameter if operation has to be lazily evaluated self._trigger: Trigger | None - if operation and (iscoroutinefunction(operation['fn']) or inspect.isgeneratorfunction(operation['fn'])): + if operation and _is_async_operation(operation): self._trigger = Trigger(internal=True) self._current_ = Undefined self._dirty = True # Otherwise current will be stuck as Undefined. @@ -3031,12 +3061,7 @@ def _obj_dirty(self) -> bool: def _is_async(self) -> bool: if not self._operation: return False - fn = self._operation["fn"] - return ( - inspect.iscoroutinefunction(fn) or - inspect.isasyncgenfunction(fn) or - inspect.isgeneratorfunction(fn) - ) + return _is_async_operation(self._operation) @property def _awaiting(self) -> bool: @@ -3056,10 +3081,11 @@ 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. + A coroutine or generator function declared a reference with + ``param.bind`` and 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 @@ -3884,6 +3910,8 @@ def _clone(self, operation=None, copy=False, **kwargs) -> Self: kwargs = dict(prev=self, **dict(self._kwargs, **kwargs)) kwargs = t.cast('dict[str, t.Any]', dict(self._display_opts, **kwargs)) error_mode = t.cast('str', kwargs.pop('error_mode', self._error_mode)) + if not reuse and operation is not None: + _warn_bare_async_inputs(*operation.get('args', ()), *(operation.get('kwargs') or {}).values()) if not reuse and error_mode == 'raise' and operation is not None: inputs = (*operation.get('args', ()), *(operation.get('kwargs') or {}).values()) if any(isinstance(arg, rx) and arg._error_mode == 'propagate' for arg in inputs): @@ -3996,6 +4024,11 @@ def __call__(self, *args, **kwargs): if method in rx._method_handlers: handler = rx._method_handlers[method] method = handler(self) + callee = None + elif new._method: + callee = getattr(new._current_, new._method, None) + else: + callee = new._current_ new._method = None kwargs = dict(kwargs) operation = { @@ -4005,6 +4038,8 @@ def __call__(self, *args, **kwargs): 'reverse': False, 'process_failures': self._process_failures, } + if _is_async_fn(callee): + operation['async'] = True return new._clone(operation) #---------------------------------------------------------------- @@ -4199,7 +4234,7 @@ def _resolve_input(self, arg): return value if any(ref._settling for ref in _iter_rx(arg)): raise Skip - val = resolve_value(arg) + val = _resolve_operand(arg) if val is Skip or val is Undefined: raise Skip return val @@ -4357,6 +4392,79 @@ def _iter_rx(value: t.Any) -> Iterator[rx]: yield from _iter_rx(v) +if sys.version_info >= (3, 14): + import annotationlib + + # Annotations are irrelevant here and evaluating them can raise, e.g. on + # names only imported under TYPE_CHECKING. + _signature = partial(inspect.signature, annotation_format=annotationlib.Format.FORWARDREF) +else: + _signature = inspect.signature + + +def _is_async_operation(operation: dict[str, t.Any]) -> bool: + """ + Whether an operation produces an awaitable or (async) generator, either + because its fn is async or because it calls an async function. + """ + return operation.get('async', False) or _is_async_fn(operation['fn']) + + +def _held_as_value(fn: t.Any) -> bool: + """ + Whether rx keeps a bare coroutine or generator function as a value, + because it cannot be called without arguments to resolve it. + """ + if hasattr(fn, '_dinfo') or not _is_async_fn(fn): + return False + try: + sig = _signature(fn) + except (TypeError, ValueError): + return False + return any( + p.default is p.empty and p.kind not in (p.VAR_POSITIONAL, p.VAR_KEYWORD) + for p in sig.parameters.values() + ) + + +def _resolve_operand(value: t.Any) -> t.Any: + """ + ``resolve_value`` for rx inputs, keeping functions that require + arguments as values instead of calling them. + """ + if isinstance(value, (list, tuple)): + return type(value)(_resolve_operand(v) for v in value) + elif isinstance(value, dict): + return type(value)((_resolve_operand(k), _resolve_operand(v)) for k, v in value.items()) + elif isinstance(value, slice): + return slice( + _resolve_operand(value.start), + _resolve_operand(value.stop), + _resolve_operand(value.step) + ) + if _held_as_value(transform_reference(value)): + return value + return resolve_value(value, recursive=False) + + +def _warn_bare_async_inputs(*values: t.Any) -> None: + """ + Warn about bare coroutine and generator functions an rx operation would + call, descending into the containers ``resolve_value`` descends into. + """ + for value in values: + if isinstance(value, rx): + continue + elif isinstance(value, (list, tuple)): + _warn_bare_async_inputs(*value) + elif isinstance(value, dict): + _warn_bare_async_inputs(*value.keys(), *value.values()) + elif isinstance(value, slice): + _warn_bare_async_inputs(value.start, value.stop, value.step) + elif _is_bare_async_ref(value := transform_reference(value)) and not _held_as_value(value): + _warn_bare_async_ref(value, operand=True) + + def _iter_bare_params(value: t.Any) -> Iterator[Parameter]: """ Yield ``Parameter`` objects nested anywhere inside an operation diff --git a/tests/testreactive.py b/tests/testreactive.py index 81203084f..5c56c67b5 100644 --- a/tests/testreactive.py +++ b/tests/testreactive.py @@ -1117,6 +1117,77 @@ async def async_func(value): await async_wait_until(lambda: expr.rx.value == 2) assert not expr.rx.awaiting + async def test_reactive_call_async_function(self): + async def async_func(value): + await asyncio.sleep(0.02) + return value + 2 + + number = rx(0) + expr = rx(async_func)(number) + + assert expr.rx.value is param.Undefined + assert expr.rx.awaiting + await async_wait_until(lambda: expr.rx.value == 2) + assert not expr.rx.awaiting + + number.rx.value = 3 + assert expr.rx.value == 2 + assert expr.rx.awaiting + await async_wait_until(lambda: expr.rx.value == 5) + + async def test_reactive_call_async_function_downstream(self): + async def async_func(value): + await asyncio.sleep(0.02) + return value + 2 + + expr = rx(async_func)(1) * 10 + + assert expr.rx.value is param.Undefined + await async_wait_until(lambda: expr.rx.value == 30) + + async def test_reactive_call_async_gen_function(self): + async def async_gen(value): + yield value + await asyncio.sleep(0.02) + yield value + 1 + + expr = rx(async_gen)(1) + expr.rx.value + + await async_wait_until(lambda: expr.rx.value == 2) + assert not expr.rx.awaiting + + async def test_reactive_call_gen_function(self): + def gen(value): + yield value + yield value + 1 + + expr = rx(gen)(1) + expr.rx.value + + await async_wait_until(lambda: expr.rx.value == 2) + + async def test_reactive_call_async_method(self): + class Fetcher: + async def fetch(self, value): + await asyncio.sleep(0.02) + return f'fetched {value}' + + number = rx(1) + expr = rx(Fetcher()).fetch(number) + + assert expr.rx.value is param.Undefined + await async_wait_until(lambda: expr.rx.value == 'fetched 1') + + number.rx.value = 2 + await async_wait_until(lambda: expr.rx.value == 'fetched 2') + + def test_reactive_call_sync_function_resolves_synchronously(self): + expr = rx(lambda value: value + 2)(rx(0)) + + assert expr.rx.value == 2 + assert not expr.rx.awaiting + async def test_reactive_awaiting_not_scheduled_until_requested(self): """ Reading awaiting must not itself schedule a resolution, so an expression @@ -1149,7 +1220,7 @@ async def async_func(): await asyncio.sleep(0.02) return {'value': 42} - expr = rx(async_func)['value'] + expr = rx(bind(async_func))['value'] assert expr.rx.value is param.Undefined assert expr.rx.awaiting @@ -1161,7 +1232,7 @@ async def async_func(): await asyncio.sleep(0.02) return {'value': 42} - source = rx(async_func) + source = rx(bind(async_func)) expr = source[rx('value')] assert expr.rx.value is param.Undefined @@ -1276,7 +1347,7 @@ async def async_func(): await asyncio.sleep(0.02) return 2 - expr = rx(async_func) + 2 + expr = rx(bind(async_func)) + 2 assert expr.rx.value is param.Undefined assert expr.rx.awaiting @@ -1305,7 +1376,7 @@ def gen(): time.sleep(0.1) yield 2 - expr = rx(gen) + 100 + expr = rx(bind(gen)) + 100 assert expr.rx.value is param.Undefined assert expr.rx.awaiting @@ -1355,7 +1426,7 @@ def gen(): return yield # pragma: no cover - expr = rx(gen) + 100 + expr = rx(bind(gen)) + 100 assert expr.rx.value is param.Undefined assert expr.rx.awaiting @@ -3062,7 +3133,7 @@ async def async_func(): await asyncio.sleep(0.02) return 2 - async_rx = rx(async_func, lazy=lazy) + 2 + async_rx = rx(bind(async_func), lazy=lazy) + 2 assert async_rx.rx.value is param.Undefined await async_wait_until(lambda: async_rx.rx.value == 4) @@ -3082,7 +3153,7 @@ def gen(): time.sleep(0.05) yield 2 - rxgen = rx(gen) + rxgen = rx(bind(gen)) assert rxgen.rx.value is param.Undefined await async_wait_until(lambda: rxgen.rx.value == 1, interval=10) await async_wait_until(lambda: rxgen.rx.value == 2) @@ -3097,7 +3168,7 @@ def gen(): time.sleep(0.05) yield 2 - rxgen = rx(gen) + rxgen = rx(bind(gen)) assert rxgen.rx.value is param.Undefined await async_wait_until(lambda: rxgen.rx.value == 1, interval=10) await asyncio.sleep(0.1) @@ -3177,7 +3248,7 @@ async def gen(): await asyncio.sleep(0.05) yield 2 - rxgen = rx(gen) + rxgen = rx(bind(gen)) assert rxgen.rx.value is param.Undefined await async_wait_until(lambda: rxgen.rx.value == 1, interval=10) await async_wait_until(lambda: rxgen.rx.value == 2) @@ -3192,7 +3263,7 @@ async def gen(): await asyncio.sleep(0.05) yield 2 - rxgen = rx(gen) + rxgen = rx(bind(gen)) assert rxgen.rx.value is param.Undefined await async_wait_until(lambda: rxgen.rx.value == 1, interval=10) await asyncio.sleep(0.1) diff --git a/tests/testrefs.py b/tests/testrefs.py index 7402f6503..b5b783ac1 100644 --- a/tests/testrefs.py +++ b/tests/testrefs.py @@ -1,10 +1,15 @@ import asyncio +import sys import threading import time +import warnings + +from functools import partial import param import pytest +from param._utils import ParamFutureWarning from param.parameterized import Skip, resolve_ref from param.reactive import bind, rx @@ -265,6 +270,174 @@ async def gen_strings(): await asyncio.sleep(0.1) assert p.string == 'string!' +async def _coroutine_string(): + return 'string!' + +async def _async_gen_string(): + yield 'string!' + +def _gen_string(): + yield 'string!' + +async def _coroutine_concat(a, b): + return a + b + +BARE_ASYNC_REFS = [ + pytest.param(_coroutine_string, r"param\.bind\(_coroutine_string\)", id='coroutine'), + pytest.param(_async_gen_string, r"param\.bind\(_async_gen_string\)", id='async_gen'), + pytest.param(_gen_string, r"param\.bind\(_gen_string\)", id='gen'), + pytest.param( + partial(_coroutine_concat, 'string', '!'), + r"param\.bind\(_coroutine_concat, \.\.\.\)", id='partial' + ), +] + +BARE_ASYNC_FUNCTIONS = [pytest.param(p.values[0], id=p.id) for p in BARE_ASYNC_REFS] + +@pytest.mark.parametrize('fn', BARE_ASYNC_FUNCTIONS) +async def test_bare_async_ref_resolves_on_init_without_warning(fn): + with warnings.catch_warnings(): + warnings.simplefilter('error', ParamFutureWarning) + p = Parameters(string=fn) + await wait_for_value(p, 'string', 'string!') + +@pytest.mark.parametrize('fn', BARE_ASYNC_FUNCTIONS) +async def test_bare_async_ref_resolves_on_set_without_warning(fn): + p = Parameters() + with warnings.catch_warnings(): + warnings.simplefilter('error', ParamFutureWarning) + p.string = fn + await wait_for_value(p, 'string', 'string!') + +@pytest.mark.parametrize('fn, match', BARE_ASYNC_REFS) +async def test_bare_async_ref_warns_in_rx(fn, match): + with pytest.warns(ParamFutureWarning, match=match): + expr = rx(fn) + await wait_for_value(expr.rx, 'value', 'string!') + +@pytest.mark.parametrize('fn, match', BARE_ASYNC_REFS) +async def test_bound_async_ref_in_rx_does_not_warn(fn, match): + with warnings.catch_warnings(): + warnings.simplefilter('error', ParamFutureWarning) + expr = rx(bind(fn)) + await wait_for_value(expr.rx, 'value', 'string!') + +@pytest.mark.parametrize('fn', BARE_ASYNC_FUNCTIONS) +async def test_bare_async_function_on_non_ref_parameter_is_kept(fn): + class NoRefs(param.Parameterized): + value = param.Parameter() + + p = NoRefs(value=fn) + await asyncio.sleep(0.1) + assert p.value is fn + +@pytest.mark.parametrize('resolve', [ + pytest.param(lambda fn: rx(1).rx.pipe(lambda a, b: b, fn), id='pipe_arg'), + pytest.param(lambda fn: rx([1]) + [fn], id='operand'), + pytest.param(lambda fn: rx(True).rx.where(fn, 0), id='where'), + pytest.param(lambda fn: setattr(rx(1).rx, 'value', fn), id='set_value'), + pytest.param(lambda fn: param.reactive.collect(fn), id='collect'), + pytest.param( + lambda fn: rx(1).rx.pipe(lambda a, b: b, rx(2)).rx.overrides.args.__setitem__(1, fn), + id='override' + ), +]) +def test_bare_async_ref_warns_in_rx_operations(resolve): + with pytest.warns(ParamFutureWarning, match="function itself will be passed on") as record: + resolve(_gen_string) + assert not any('param.bind' in str(w.message) for w in record) + +def test_bare_async_function_in_set_does_not_warn(): + with warnings.catch_warnings(): + warnings.simplefilter('error', ParamFutureWarning) + expr = rx(1).rx.pipe(lambda a, b: b, {_gen_string}) + assert expr.rx.value == {_gen_string} + +@pytest.mark.skipif(sys.version_info < (3, 14), reason='Annotations are evaluated lazily from Python 3.14') +async def test_rx_async_function_with_unresolvable_annotation(): + async def fetch() -> Unresolvable: # noqa: F821 + return 'string!' + + with pytest.warns(ParamFutureWarning, match=r"param\.bind\(fetch\)"): + expr = rx(fetch) + await wait_for_value(expr.rx, 'value', 'string!') + +def test_rx_operation_warns_once(): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always', ParamFutureWarning) + source = rx(1) + expr = source.rx.pipe(lambda a, b: a, _gen_string) + source.rx.value = 2 + assert expr.rx.value == 2 + assert sum(issubclass(w.category, ParamFutureWarning) for w in caught) == 1 + +async def _coroutine_echo(value): + return value + +async def _async_gen_echo(value): + yield value + +def _gen_echo(value): + yield value + +ARG_ASYNC_FUNCTIONS = [ + pytest.param(_coroutine_echo, id='coroutine'), + pytest.param(_async_gen_echo, id='async_gen'), + pytest.param(_gen_echo, id='gen'), + pytest.param(partial(_coroutine_concat, 'string'), id='partial'), +] + +@pytest.mark.parametrize('fn', ARG_ASYNC_FUNCTIONS) +async def test_rx_keeps_async_function_requiring_args_as_value(fn): + with warnings.catch_warnings(): + warnings.simplefilter('error', ParamFutureWarning) + expr = rx(fn) + await asyncio.sleep(0.05) + assert expr.rx.value is fn + +@pytest.mark.parametrize('fn', ARG_ASYNC_FUNCTIONS) +@pytest.mark.parametrize('build, expected', [ + pytest.param(lambda fn: rx(1).rx.pipe(lambda a, b: b, fn), lambda fn: fn, id='pipe_arg'), + pytest.param(lambda fn: rx([1]) + [fn], lambda fn: [1, fn], id='operand'), + pytest.param(lambda fn: rx(True).rx.where(fn, 0), lambda fn: fn, id='where'), +]) +async def test_rx_operation_keeps_async_function_requiring_args_as_value(fn, build, expected): + with warnings.catch_warnings(): + warnings.simplefilter('error', ParamFutureWarning) + expr = build(fn) + await asyncio.sleep(0.05) + assert expr.rx.value == expected(fn) + +@pytest.mark.parametrize('fn', ARG_ASYNC_FUNCTIONS) +async def test_rx_value_set_keeps_async_function_requiring_args(fn): + expr = rx(1) + with warnings.catch_warnings(): + warnings.simplefilter('error', ParamFutureWarning) + expr.rx.value = fn + assert expr.rx.value is fn + +@pytest.mark.parametrize('fn', ARG_ASYNC_FUNCTIONS) +async def test_rx_value_set_on_async_root_keeps_async_function_requiring_args(fn): + expr = rx(bind(_gen_string)) + await wait_for_value(expr.rx, 'value', 'string!') + with warnings.catch_warnings(): + warnings.simplefilter('error', ParamFutureWarning) + expr.rx.value = fn + await asyncio.sleep(0.05) + assert expr.rx.value is fn + +async def test_rx_resolves_async_function_with_only_default_args(): + async def fetch(value='string!'): + return value + + with pytest.warns(ParamFutureWarning, match=r"param\.bind\(fetch\)"): + expr = rx(fetch) + await wait_for_value(expr.rx, 'value', 'string!') + +def test_parameter_still_calls_async_function_requiring_args(): + with pytest.raises(TypeError, match="missing 1 required positional argument"): + Parameters(string=_gen_echo) + async def test_async_bind_ref(): p = Parameters()