Dataclass Descriptor#

class SignalGroupDescriptor:
    """Create a [`psygnal.SignalGroup`][] on first instance attribute access.

    This descriptor is designed to be used as a class attribute on a dataclass-like
    class (e.g. a [`dataclass`](https://docs.python.org/3/library/dataclasses.html), a
    [`pydantic.BaseModel`](https://docs.pydantic.dev/usage/models/), an
    [attrs](https://www.attrs.org/en/stable/overview.html) class, a
    [`msgspec.Struct`](https://jcristharif.com/msgspec/structs.html)) On first access of
    the descriptor on an instance, it will create a [`SignalGroup`][psygnal.SignalGroup]
    bound to the instance, with a [`SignalInstance`][psygnal.SignalInstance] for each
    field in the dataclass.

    !!!important
        Using this descriptor will *patch* the class's `__setattr__` method to emit
        events when fields change. (That patching occurs on first access of the
        descriptor name on an instance).  To prevent this patching, you can set
        `patch_setattr=False` when creating the descriptor, but then you will need to
        manually call `emit` on the appropriate `SignalInstance` when you want to emit
        an event.  Or you can use `evented_setattr` yourself

        ```python
        from psygnal._group_descriptor import evented_setattr
        from psygnal import SignalGroupDescriptor
        from dataclasses import dataclass
        from typing import ClassVar


        @dataclass
        class Foo:
            x: int
            _events: ClassVar = SignalGroupDescriptor(patch_setattr=False)

            @evented_setattr("_events")  # pass the name of your SignalGroup
            def __setattr__(self, name: str, value: Any) -> None:
                super().__setattr__(name, value)
        ```

        *This currently requires a private import, please open an issue if you would
        like to depend on this functionality.*

    Parameters
    ----------
    equality_operators : dict[str, Callable[[Any, Any], bool]], optional
        A dictionary mapping field names to custom equality operators, where an equality
        operator is a callable that accepts two arguments and returns True if the two
        objects are equal. This will be used when comparing the old and new values of a
        field to determine whether to emit an event. If not provided, the default
        equality operator is `operator.eq`, except for numpy arrays, where
        `np.array_equal` is used.
    warn_on_no_fields : bool, optional
        If `True` (the default), a warning will be emitted if no mutable dataclass-like
        fields are found on the object.
    cache_on_instance : bool, optional
        If `True` (the default), a newly-created SignalGroup instance will be cached on
        the instance itself, so that subsequent accesses to the descriptor will return
        the same SignalGroup instance.  This makes for slightly faster subsequent
        access, but means that the owner instance will no longer be pickleable.  If
        `False`, the SignalGroup instance will *still* be cached, but not on the
        instance itself.
    patch_setattr : bool, optional
        If `True` (the default), a new `__setattr__` method will be created that emits
        events when fields change.  If `False`, no `__setattr__` method will be
        created.  (This will prevent signal emission, and assumes you are using a
        different mechanism to emit signals when fields change.)
    signal_group_class : type[SignalGroup] | None, optional
        A custom SignalGroup class to use, SignalGroup if None, by default None
    collect_fields : bool, optional
        Create a signal for each field in the dataclass. If True, the `SignalGroup`
        instance will be a subclass of `signal_group_class` (SignalGroup if it is None).
        If False, a deepcopy of `signal_group_class` will be used.
        Default to True
    signal_aliases: Mapping[str, str | None] | Callable[[str], str | None] | None
        If defined, a mapping between field name and signal name. Field names that are
        not `signal_aliases` keys are not aliased (the signal name is the field name).
        If the dict value is None, do not create a signal associated with this field.
        If a callable, the signal name is the output of the function applied to the
        field name. If the output is None, no signal is created for this field.
        If None, defaults to an empty dict, no aliases.
        Default to None

    Examples
    --------
    ```python
    from typing import ClassVar
    from dataclasses import dataclass
    from psygnal import SignalGroupDescriptor


    @dataclass
    class Person:
        name: str
        age: int = 0
        events: ClassVar[SignalGroupDescriptor] = SignalGroupDescriptor()


    john = Person("John", 40)
    john.events.age.connect(print)
    john.age += 1  # prints 41
    ```
    """

    # map of id(obj) -> SignalGroup
    # cached here in case the object isn't modifiable
    _instance_map: ClassVar[dict[int, SignalGroup]] = {}

    def __init__(
        self,
        *,
        equality_operators: dict[str, EqOperator] | None = None,
        warn_on_no_fields: bool = True,
        cache_on_instance: bool = True,
        patch_setattr: bool = True,
        signal_group_class: type[SignalGroup] | None = None,
        collect_fields: bool = True,
        signal_aliases: Mapping[str, str | None] | FieldAliasFunc | None = None,
    ):
        grp_cls = signal_group_class or SignalGroup
        if not (isinstance(grp_cls, type) and issubclass(grp_cls, SignalGroup)):
            raise TypeError(  # pragma: no cover
                f"'signal_group_class' must be a subclass of SignalGroup, "
                f"not {grp_cls}"
            )
        if not collect_fields:
            if grp_cls is SignalGroup:
                raise ValueError(
                    "Cannot use SignalGroup with `collect_fields=False`. "
                    "Use a custom SignalGroup subclass instead."
                )

            if callable(signal_aliases):
                raise ValueError(
                    "Cannot use a Callable for `signal_aliases` with "
                    "`collect_fields=False`"
                )

        self._name: str | None = None
        self._eqop = tuple(equality_operators.items()) if equality_operators else None
        self._warn_on_no_fields = warn_on_no_fields
        self._cache_on_instance = cache_on_instance
        self._patch_setattr = patch_setattr
        self._signal_group_class: type[SignalGroup] = grp_cls
        self._collect_fields = collect_fields
        self._signal_aliases = signal_aliases

        self._signal_groups: dict[int, type[SignalGroup]] = {}

    def __set_name__(self, owner: type, name: str) -> None:
        """Called when this descriptor is added to class `owner` as attribute `name`."""
        self._name = name
        with contextlib.suppress(AttributeError):
            # This is the flag that identifies this object as evented
            setattr(owner, PSYGNAL_GROUP_NAME, name)

    def _do_patch_setattr(self, owner: type, with_aliases: bool = True) -> None:
        """Patch the owner class's __setattr__ method to emit events."""
        if not self._patch_setattr:
            return
        if getattr(owner.__setattr__, PATCHED_BY_PSYGNAL, False):
            return

        name = self._name
        if not (name and hasattr(owner, name)):  # pragma: no cover
            # this should never happen... but if it does, we'll get errors
            # every time we set an attribute on the class.  So raise now.
            raise AttributeError("SignalGroupDescriptor has not been set on the class")

        try:
            # assign a new __setattr__ method to the class
            owner.__setattr__ = evented_setattr(  # type: ignore
                name,
                owner.__setattr__,  # type: ignore
                with_aliases=with_aliases,
            )
        except Exception as e:  # pragma: no cover
            # not sure what might cause this ... but it will have consequences
            raise type(e)(
                f"Could not update __setattr__ on class: {owner}. Events will not be "
                "emitted when fields change."
            ) from e

    @overload
    def __get__(self, instance: None, owner: type) -> SignalGroupDescriptor: ...

    @overload
    def __get__(self, instance: object, owner: type) -> SignalGroup: ...

    def __get__(
        self, instance: object, owner: type
    ) -> SignalGroup | SignalGroupDescriptor:
        """Return a SignalGroup instance for `instance`."""
        if instance is None:
            return self

        signal_group = self._get_signal_group(owner)

        # if we haven't yet instantiated a SignalGroup for this instance,
        # do it now and cache it.  Note that we cache it here in addition to
        # the instance (in case the instance is not modifiable).
        obj_id = id(instance)
        if obj_id not in self._instance_map:
            # cache it
            self._instance_map[obj_id] = signal_group(instance)
            # also *try* to set it on the instance as well, since it will skip all the
            # __get__ logic in the future, but if it fails, no big deal.
            if self._name and self._cache_on_instance:
                with contextlib.suppress(Exception):
                    setattr(instance, self._name, self._instance_map[obj_id])

            # clean up the cache when the instance is deleted
            with contextlib.suppress(TypeError):  # if it's not weakref-able
                weakref.finalize(instance, self._instance_map.pop, obj_id, None)

        return self._instance_map[obj_id]

    def _get_signal_group(self, owner: type) -> type[SignalGroup]:
        type_id = id(owner)
        if type_id not in self._signal_groups:
            self._signal_groups[type_id] = self._create_group(owner)
        return self._signal_groups[type_id]

    def _create_group(self, owner: type) -> type[SignalGroup]:
        if not self._collect_fields:
            # Do not collect fields from owner class
            Group = copy.deepcopy(self._signal_group_class)

            # Add aliases
            if isinstance(self._signal_aliases, dict):
                Group._psygnal_aliases.update(self._signal_aliases)

        else:
            # Collect fields and create SignalGroup subclass
            Group = _build_dataclass_signal_group(
                owner,
                self._signal_group_class,
                equality_operators=self._eqop,
                signal_aliases=self._signal_aliases,
            )

        if self._warn_on_no_fields and not Group._psygnal_signals:
            warnings.warn(
                f"No mutable fields found on class {owner}: no events will be "
                "emitted. (Is this a dataclass, attrs, msgspec, or pydantic model?)",
                stacklevel=2,
            )

        self._do_patch_setattr(owner, with_aliases=bool(Group._psygnal_aliases))
        return Group