diff --git a/psygnal/_signal.py b/psygnal/_signal.py index 23851323..33360224 100644 --- a/psygnal/_signal.py +++ b/psygnal/_signal.py @@ -26,7 +26,7 @@ from typing_extensions import Literal, get_args, get_origin, get_type_hints -MethodRef = Tuple["weakref.ReferenceType[object]", Union[Callable, str]] +MethodRef = Tuple["weakref.ReferenceType[object]", str, Optional[Callable]] NormedCallback = Union[MethodRef, Callable] StoredSlot = Tuple[NormedCallback, Optional[int]] AnyType = Type[Any] @@ -329,7 +329,6 @@ def connect( ) -> Callable: ... # pragma: no cover - # TODO: allow connect as decorator with arguments def connect( self, slot: Optional[Callable] = None, @@ -411,6 +410,88 @@ def _wrapper(slot: Callable) -> Callable: return _wrapper(slot) if slot else _wrapper + def connect_setattr( + self, + obj: Union[weakref.ref, object], + attr: str, + maxargs: Optional[int] = None, + ) -> MethodRef: + """Bind the an object attribute to emitted value of this signal. + + Equivalent to calling self.connect(partial(setattr, obj, attr)), with + weakref safety. The return object can be used to disconnect, (or can use + disconnect_setattr). + + Parameters + ---------- + obj : Union[weakref.ref, object] + An object or weak reference to an object. + attr : str + The name of an attribute on `obj` that should be set to the value of this + signal when emitted. + maxargs : int, optional + max number of positional args to accept + + Returns + ------- + Tuple + (weakref.ref, name, callable). Reference to the object, name of the + attribute, and setattr closure. Can be used to disconnect the slot. + + Raises + ------ + ValueError + If this is not a single-value signal + AttributeError + If `obj` has no attribute `attr`. + """ + ref = obj if isinstance(obj, weakref.ref) else weakref.ref(obj) + if not hasattr(ref(), attr): + raise AttributeError(f"Object {ref()} has no attribute {attr!r}") + + with self._lock: + + def _slot(*args: Any) -> None: + setattr(ref(), attr, args[0] if len(args) == 1 else args) + + normed_callback = (ref, attr, _slot) + self._slots.append((normed_callback, maxargs)) + return normed_callback + + def disconnect_setattr( + self, obj: object, attr: str, missing_ok: bool = True + ) -> None: + """Disconnect a previously connected attribute setter. + + Parameters + ---------- + obj : object + An object. + attr : str + The name of an attribute on `obj` that was previously used for + `connect_setattr`. + missing_ok : bool, optional + If `False` and the provided `slot` is not connected, raises `ValueError. + by default `True` + + Raises + ------ + ValueError + If `missing_ok` is True and no attribute setter is connected. + """ + with self._lock: + idx = None + for i, (slot, _) in enumerate(self._slots): + if isinstance(slot, tuple): + ref, name, _ = slot + if ref() is obj and attr == name: + idx = i + break + if idx is not None: + self._slots.pop(idx) + elif not missing_ok: + raise ValueError(f"No attribute setter connected for {obj}.{attr}") + def _check_nargs( self, slot: Callable, spec: Signature ) -> Tuple[Optional[Signature], Optional[int]]: @@ -446,11 +527,11 @@ def _raise_connection_error(self, slot: Callable, extra: str = "") -> NoReturn: def _normalize_slot(self, slot: NormedCallback) -> NormedCallback: if isinstance(slot, MethodType): - return _get_proper_name(slot) + return _get_proper_name(slot) + (None,) if isinstance(slot, PartialMethod): return _partial_weakref(slot) if isinstance(slot, tuple) and not isinstance(slot[0], weakref.ref): - return (weakref.ref(slot[0]), slot[1]) + return (weakref.ref(slot[0]), slot[1], slot[2]) return slot def _slot_index(self, slot: NormedCallback) -> int: @@ -632,15 +713,15 @@ def _run_emit_loop(self, args: Tuple[Any, ...]) -> None: with Signal._emitting(self): for (slot, max_args) in self._slots: if isinstance(slot, tuple): - _ref, method = slot + _ref, name, method = slot obj = _ref() if obj is None: rem.append(slot) # add dead weakref continue - if callable(method): + if method is not None: cb = method else: - cb = getattr(obj, method, None) + cb = getattr(obj, name, None) if cb is None: # pragma: no cover rem.append(slot) # object has changed? continue @@ -928,7 +1009,7 @@ def _is_subclass(left: AnyType, right: type) -> bool: return issubclass(left, right) -def _partial_weakref(slot_partial: PartialMethod) -> Tuple[weakref.ref, Callable]: +def _partial_weakref(slot_partial: PartialMethod) -> Tuple[weakref.ref, str, Callable]: """For partial methods, make the weakref point to the wrapped object.""" ref, name = _get_proper_name(slot_partial.func) args_ = slot_partial.args @@ -937,7 +1018,7 @@ def _partial_weakref(slot_partial: PartialMethod) -> Tuple[weakref.ref, Callable def wrap(*args: Any, **kwargs: Any) -> Any: getattr(ref(), name)(*args_, *args, **kwargs_, **kwargs) - return (ref, wrap) + return (ref, name, wrap) def _get_proper_name(slot: MethodType) -> Tuple[weakref.ref, str]: diff --git a/tests/test_psygnal.py b/tests/test_psygnal.py index f78d693d..88816562 100644 --- a/tests/test_psygnal.py +++ b/tests/test_psygnal.py @@ -323,10 +323,10 @@ def test_norm_slot(): normed1 = e.one_int._normalize_slot(r.f_any) normed2 = e.one_int._normalize_slot(normed1) - normed3 = e.one_int._normalize_slot((r, "f_any")) - normed3 = e.one_int._normalize_slot((weakref.ref(r), "f_any")) - assert normed1 == (weakref.ref(r), "f_any") - assert normed1 == normed2 == normed3 + normed3 = e.one_int._normalize_slot((r, "f_any", None)) + normed4 = e.one_int._normalize_slot((weakref.ref(r), "f_any", None)) + assert normed1 == (weakref.ref(r), "f_any", None) + assert normed1 == normed2 == normed3 == normed4 assert e.one_int._normalize_slot(f_any) == f_any @@ -616,3 +616,40 @@ def test_get_proper_name(): assert _get_proper_name(obj.f_int_decorated_stupid)[1] == "f_int_decorated_stupid" assert _get_proper_name(obj.f_int_decorated_good)[1] == "f_int_decorated_good" assert _get_proper_name(obj.f_any_assigned)[1] == "f_any_assigned" + + +def test_property_connect(): + class A: + def __init__(self): + self.li = [] + + @property + def x(self): + return self.li + + @x.setter + def x(self, value): + self.li.append(value) + + a = A() + emitter = Emitter() + emitter.one_int.connect_setattr(a, "x") + assert len(emitter.one_int) == 1 + emitter.two_int.connect_setattr(a, "x") + assert len(emitter.two_int) == 1 + emitter.one_int.emit(1) + assert a.li == [1] + emitter.two_int.emit(1, 1) + assert a.li == [1, (1, 1)] + emitter.two_int.disconnect_setattr(a, "x") + assert len(emitter.two_int) == 0 + with pytest.raises(ValueError): + emitter.two_int.disconnect_setattr(a, "x", missing_ok=False) + emitter.two_int.disconnect_setattr(a, "x") + s = emitter.two_int.connect_setattr(a, "x", maxargs=1) + emitter.two_int.emit(2, 3) + assert a.li == [1, (1, 1), 2] + emitter.two_int.disconnect(s, missing_ok=False) + + with pytest.raises(AttributeError): + emitter.one_int.connect_setattr(a, "y")