Skip to content

Dataclass Descriptor#

These objects are used to turn dataclasses into "evented" objects, with a signal emitted whenever a field is changed.

psygnal.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, a pydantic.BaseModel, an attrs class, a msgspec.Struct) On first access of the descriptor on an instance, it will create a SignalGroup bound to the instance, with a 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

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.

  • signal_group_class (type[SignalGroup], optional) –

    A custom SignalGroup class to use, by default None

  • 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.)

Examples:

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
Source code in psygnal/_group_descriptor.py
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
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.
    signal_group_class : type[SignalGroup], optional
        A custom SignalGroup class to use, by default None
    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.)

    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
    ```
    """

    def __init__(
        self,
        *,
        equality_operators: dict[str, EqOperator] | None = None,
        signal_group_class: type[SignalGroup] | None = None,
        warn_on_no_fields: bool = True,
        cache_on_instance: bool = True,
        patch_setattr: bool = True,
    ):
        self._signal_group = signal_group_class
        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

    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) -> 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

        try:
            # assign a new __setattr__ method to the class
            owner.__setattr__ = evented_setattr(  # type: ignore
                cast(str, self._name),
                owner.__setattr__,  # type: ignore
            )
        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

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

    @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

        # 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] = self._create_group(owner)(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 _create_group(self, owner: type) -> type[SignalGroup]:
        Group = self._signal_group or _build_dataclass_signal_group(owner, self._eqop)
        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)
        return Group

psygnal.evented(cls=None, *, events_namespace='events', equality_operators=None, warn_on_no_fields=True, cache_on_instance=True) #

A decorator to add events to a dataclass.

See also the documentation for SignalGroupDescriptor. This decorator is equivalent setting a class variable named events to a new SignalGroupDescriptor instance.

Note that this decorator will modify cls in place, as well as return it.

Tip

It is recommended to use the SignalGroupDescriptor descriptor rather than the decorator, as it it is more explicit and provides for easier static type inference.

Parameters:

  • cls (type) –

    The class to decorate.

  • events_namespace (str) –

    The name of the namespace to add the events to, by default "events"

  • equality_operators (Optional[Dict[str, Callable]]) –

    A dictionary mapping field names to equality operators (a function that takes two values and returns True if they are equal). These will be used to determine if a field has changed when setting a new value. By default, this will use the __eq__ method of the field type, or np.array_equal, for numpy arrays. But you can provide your own if you want to customize how equality is checked. Alternatively, if the class has an __eq_operators__ class attribute, it will be used.

  • warn_on_no_fields (bool) –

    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.

Returns:

  • type

    The decorated class, which gains a new SignalGroup instance at the events_namespace attribute (by default, events).

Raises:

  • TypeError

    If the class is frozen or is not a class.

Examples:

from psygnal import evented
from dataclasses import dataclass

@evented
@dataclass
class Person:
    name: str
    age: int = 0
Source code in psygnal/_evented_decorator.py
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def evented(
    cls: Optional[T] = None,
    *,
    events_namespace: str = "events",
    equality_operators: Optional[Dict[str, EqOperator]] = None,
    warn_on_no_fields: bool = True,
    cache_on_instance: bool = True,
) -> Union[Callable[[T], T], T]:
    """A decorator to add events to a dataclass.

    See also the documentation for
    [`SignalGroupDescriptor`][psygnal.SignalGroupDescriptor].  This decorator is
    equivalent setting a class variable named `events` to a new
    `SignalGroupDescriptor` instance.

    Note that this decorator will modify `cls` *in place*, as well as return it.

    !!!tip
        It is recommended to use the `SignalGroupDescriptor` descriptor rather than
        the decorator, as it it is more explicit and provides for easier static type
        inference.

    Parameters
    ----------
    cls : type
        The class to decorate.
    events_namespace : str
        The name of the namespace to add the events to, by default `"events"`
    equality_operators : Optional[Dict[str, Callable]]
        A dictionary mapping field names to equality operators (a function that takes
        two values and returns `True` if they are equal). These will be used to
        determine if a field has changed when setting a new value.  By default, this
        will use the `__eq__` method of the field type, or np.array_equal, for numpy
        arrays.  But you can provide your own if you want to customize how equality is
        checked. Alternatively, if the class has an `__eq_operators__` class attribute,
        it will be used.
    warn_on_no_fields : bool
        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.

    Returns
    -------
    type
        The decorated class, which gains a new SignalGroup instance at the
        `events_namespace` attribute (by default, `events`).

    Raises
    ------
    TypeError
        If the class is frozen or is not a class.

    Examples
    --------
    ```python
    from psygnal import evented
    from dataclasses import dataclass

    @evented
    @dataclass
    class Person:
        name: str
        age: int = 0
    ```
    """

    def _decorate(cls: T) -> T:
        if not isinstance(cls, type):  # pragma: no cover
            raise TypeError("evented can only be used on classes")

        descriptor = SignalGroupDescriptor(
            equality_operators=equality_operators,
            warn_on_no_fields=warn_on_no_fields,
            cache_on_instance=cache_on_instance,
        )
        # as a decorator, this will have already been called
        descriptor.__set_name__(cls, events_namespace)
        setattr(cls, events_namespace, descriptor)
        return cls

    return _decorate(cls) if cls is not None else _decorate