Skip to content

Evented Containers#

These classes provide "evented" versions of mutable python containers. They each have an events attribute (SignalGroup) that has a variety of signals that will emit whenever the container is mutated. See Container SignalGroups for the corresponding container type for details on the available signals.

psygnal.containers.EventedDict #

Bases: TypedMutableMapping[_K, _V]

Mutable mapping that emits events when altered.

This class is designed to behave exactly like the builtin dict, but will emit events before and after all mutations (addition, removal, and changing).

Parameters:

  • data (Union[Mapping[_K, _V], Iterable[Tuple[_K, _V]], None], optional) –

    Data suitable of passing to dict(). Mapping of {key: value} pairs, or Iterable of two-tuples [(key, value), ...], or None to create an

  • basetype (TypeOrSequenceOfTypes, optional) –

    Type or Sequence of Type objects. If provided, values entered into this Mapping must be an instance of one of the provided types. by default ().

Attributes:

  • events (DictEvents) –

    The SignalGroup object that emits all events available on an EventedDict.

Source code in psygnal/containers/_evented_dict.py
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
class EventedDict(TypedMutableMapping[_K, _V]):
    """Mutable mapping that emits events when altered.

    This class is designed to behave exactly like the builtin [`dict`][], but
    will emit events before and after all mutations (addition, removal, and
    changing).

    Parameters
    ----------
    data : Union[Mapping[_K, _V], Iterable[Tuple[_K, _V]], None], optional
        Data suitable of passing to dict(). Mapping of {key: value} pairs, or
        Iterable of two-tuples [(key, value), ...], or None to create an
    basetype : TypeOrSequenceOfTypes, optional
        Type or Sequence of Type objects. If provided, values entered into this Mapping
        must be an instance of one of the provided types. by default ().

    Attributes
    ----------
    events: DictEvents
        The `SignalGroup` object that emits all events available on an `EventedDict`.
    """

    events: DictEvents  # pragma: no cover

    def __init__(
        self,
        data: DictArg | None = None,
        *,
        basetype: TypeOrSequenceOfTypes = (),
        **kwargs: _V,
    ):
        self.events = DictEvents()
        super().__init__(data, basetype=basetype, **kwargs)

    def __setitem__(self, key: _K, value: _V) -> None:
        if key not in self._dict:
            self.events.adding.emit(key)
            super().__setitem__(key, value)
            self.events.added.emit(key, value)
        else:
            old_value = self._dict[key]
            if value is not old_value:
                self.events.changing.emit(key)
                super().__setitem__(key, value)
                self.events.changed.emit(key, old_value, value)

    def __delitem__(self, key: _K) -> None:
        item = self._dict[key]
        self.events.removing.emit(key)
        super().__delitem__(key)
        self.events.removed.emit(key, item)

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}({super().__repr__()})"

psygnal.containers.EventedList #

Bases: MutableSequence[_T]

Mutable Sequence that emits events when altered.

This class is designed to behave exactly like the builtin list, but will emit events before and after all mutations (insertion, removal, setting, and moving).

Parameters:

  • data (iterable, optional) –

    Elements to initialize the list with.

  • hashable (bool) –

    Whether the list should be hashable as id(self). By default True.

  • child_events (bool) –

    Whether to re-emit events from emitted from evented items in the list (i.e. items that have SignalInstances). If True, child events can be connected at EventedList.events.child_event. By default, False.

Attributes:

  • events (ListEvents) –

    SignalGroup that with events related to list mutation. (see ListEvents)

Source code in psygnal/containers/_evented_list.py
 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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
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
class EventedList(MutableSequence[_T]):
    """Mutable Sequence that emits events when altered.

    This class is designed to behave exactly like the builtin `list`, but
    will emit events before and after all mutations (insertion, removal,
    setting, and moving).

    Parameters
    ----------
    data : iterable, optional
        Elements to initialize the list with.
    hashable : bool
        Whether the list should be hashable as id(self). By default `True`.
    child_events: bool
        Whether to re-emit events from emitted from evented items in the list
        (i.e. items that have SignalInstances). If `True`, child events can be connected
        at `EventedList.events.child_event`. By default, `False`.

    Attributes
    ----------
    events : ListEvents
        SignalGroup that with events related to list mutation.  (see ListEvents)
    """

    events: ListEvents  # pragma: no cover

    def __init__(
        self,
        data: Iterable[_T] = (),
        *,
        hashable: bool = True,
        child_events: bool = False,
    ):
        super().__init__()
        self._data: list[_T] = []
        self._hashable = hashable
        self._child_events = child_events
        self.events = ListEvents()
        self.extend(data)

    # WAIT!! ... Read the module docstring before reimplement these methods
    # def append(self, item): ...
    # def clear(self): ...
    # def pop(self, index=-1): ...
    # def extend(self, value: Iterable[_T]): ...
    # def remove(self, value: Any): ...

    def insert(self, index: int, value: _T) -> None:
        """Insert `value` before index."""
        _value = self._pre_insert(value)
        self.events.inserting.emit(index)
        self._data.insert(index, _value)
        self.events.inserted.emit(index, value)
        self._post_insert(value)

    @overload
    def __getitem__(self, key: int) -> _T: ...

    @overload
    def __getitem__(self, key: slice) -> Self: ...

    def __getitem__(self, key: Index) -> _T | Self:
        """Return self[key]."""
        result = self._data[key]
        return self.__newlike__(result) if isinstance(result, list) else result

    @overload
    def __setitem__(self, key: int, value: _T) -> None: ...

    @overload
    def __setitem__(self, key: slice, value: Iterable[_T]) -> None: ...

    def __setitem__(self, key: Index, value: _T | Iterable[_T]) -> None:
        """Set self[key] to value."""
        old = self._data[key]
        if value is old:
            return

        # sourcery skip: hoist-similar-statement-from-if, hoist-statement-from-if
        if isinstance(key, slice):
            if not isinstance(value, Iterable):
                raise TypeError("Can only assign an iterable to slice")
            value = [self._pre_insert(v) for v in value]  # before we mutate the list
            self._data[key] = value
        else:
            value = self._pre_insert(cast("_T", value))
            self._data[key] = value

        self.events.changed.emit(key, old, value)

    def __delitem__(self, key: Index) -> None:
        """Delete self[key]."""
        # delete from the end
        for parent, index in sorted(self._delitem_indices(key), reverse=True):
            parent.events.removing.emit(index)
            parent._pre_remove(index)
            item = parent._data.pop(index)
            self.events.removed.emit(index, item)

    def _delitem_indices(self, key: Index) -> Iterable[tuple[EventedList[_T], int]]:
        # returning (self, int) allows subclasses to pass nested members
        if isinstance(key, int):
            yield (self, key if key >= 0 else key + len(self))
        elif isinstance(key, slice):
            yield from ((self, i) for i in range(*key.indices(len(self))))
        else:
            n = repr(type(key).__name__)
            raise TypeError(f"EventedList indices must be integers or slices, not {n}")

    def _pre_insert(self, value: _T) -> _T:
        """Validate and or modify values prior to inserted."""
        return value

    def _post_insert(self, new_item: _T) -> None:
        """Modify and or handle values after insertion."""
        if self._child_events:
            self._connect_child_emitters(new_item)

    def _pre_remove(self, index: int) -> None:
        """Modify and or handle values before removal."""
        if self._child_events:
            self._disconnect_child_emitters(self[index])

    def __newlike__(self, iterable: Iterable[_T]) -> Self:
        """Return new instance of same class."""
        return self.__class__(iterable)

    def copy(self) -> Self:
        """Return a shallow copy of the list."""
        return self.__newlike__(self)

    def __copy__(self) -> Self:
        return self.copy()

    def __add__(self, other: Iterable[_T]) -> Self:
        """Add other to self, return new object."""
        copy = self.copy()
        copy.extend(other)
        return copy

    def __iadd__(self, other: Iterable[_T]) -> Self:
        """Add other to self in place (self += other)."""
        self.extend(other)
        return self

    def __radd__(self, other: list) -> list:
        """Reflected add (other + self).  Cast self to list."""
        return other + list(self)

    def __len__(self) -> int:
        """Return len(self)."""
        return len(self._data)

    def __repr__(self) -> str:
        """Return repr(self)."""
        return f"{type(self).__name__}({self._data})"

    def __eq__(self, other: Any) -> bool:
        """Return self==value."""
        return bool(self._data == other)

    def __hash__(self) -> int:
        """Return hash(self)."""
        # it's important to add this to allow this object to be hashable
        # given that we've also reimplemented __eq__
        if self._hashable:
            return id(self)
        name = self.__class__.__name__
        raise TypeError(
            f"unhashable type: {name!r}. "
            f"Create with {name}(..., hashable=True) if you need hashability"
        )

    def reverse(self, *, emit_individual_events: bool = False) -> None:
        """Reverse list *IN PLACE*."""
        if emit_individual_events:
            super().reverse()
        else:
            self._data.reverse()
        self.events.reordered.emit()

    def move(self, src_index: int, dest_index: int = 0) -> bool:
        """Insert object at `src_index` before `dest_index`.

        Both indices refer to the list prior to any object removal
        (pre-move space).
        """
        if dest_index < 0:
            dest_index += len(self) + 1
        if dest_index in (src_index, src_index + 1):
            # this is a no-op
            return False

        self.events.moving.emit(src_index, dest_index)
        item = self._data.pop(src_index)
        if dest_index > src_index:
            dest_index -= 1
        self._data.insert(dest_index, item)
        self.events.moved.emit(src_index, dest_index, item)
        self.events.reordered.emit()
        return True

    def move_multiple(self, sources: Iterable[Index], dest_index: int = 0) -> int:
        """Move a batch of `sources` indices, to a single destination.

        Note, if `dest_index` is higher than any of the `sources`, then
        the resulting position of the moved objects after the move operation
        is complete will be lower than `dest_index`.

        Parameters
        ----------
        sources : Iterable[Union[int, slice]]
            A sequence of indices
        dest_index : int, optional
            The destination index.  All sources will be inserted before this
            index (in pre-move space), by default 0... which has the effect of
            "bringing to front" everything in `sources`, or acting as a
            "reorder" method if `sources` contains all indices.

        Returns
        -------
        int
            The number of successful move operations completed.

        Raises
        ------
        TypeError
            If the destination index is a slice, or any of the source indices
            are not `int` or `slice`.
        """
        # calling list here makes sure that there are no index errors up front
        move_plan = list(self._move_plan(sources, dest_index))

        # don't assume index adjacency ... so move objects one at a time
        # this *could* be simplified with an intermediate list ... but this way
        # allows any views (such as QtViews) to update themselves more easily.
        # If this needs to be changed in the future for performance reasons,
        # then the associated QtListView will need to changed from using
        # `beginMoveRows` & `endMoveRows` to using `layoutAboutToBeChanged` &
        # `layoutChanged` while *manually* updating model indices with
        # `changePersistentIndexList`.  That becomes much harder to do with
        # nested tree-like models.
        with self.events.reordered.blocked():
            for src, dest in move_plan:
                self.move(src, dest)

        self.events.reordered.emit()
        return len(move_plan)

    def _move_plan(
        self, sources: Iterable[Index], dest_index: int
    ) -> Iterable[tuple[int, int]]:
        """Yield prepared indices for a multi-move.

        Given a set of `sources` from anywhere in the list,
        and a single `dest_index`, this function computes and yields
        `(from_index, to_index)` tuples that can be used sequentially in
        single move operations.  It keeps track of what has moved where and
        updates the source and destination indices to reflect the model at each
        point in the process.

        This is useful for a drag-drop operation with a QtModel/View.

        Parameters
        ----------
        sources : Iterable[tuple[int, ...]]
            An iterable of tuple[int] that should be moved to `dest_index`.
        dest_index : Tuple[int]
            The destination for sources.
        """
        if isinstance(dest_index, slice):
            raise TypeError("Destination index may not be a slice")  # pragma: no cover

        to_move: list[int] = []
        for idx in sources:
            if isinstance(idx, slice):
                to_move.extend(list(range(*idx.indices(len(self)))))
            elif isinstance(idx, int):
                to_move.append(idx)
            else:
                raise TypeError(
                    "Can only move integer or slice indices"
                )  # pragma: no cover

        to_move = list(dict.fromkeys(to_move))

        if dest_index < 0:
            dest_index += len(self) + 1

        d_inc = 0
        popped: list[int] = []
        for i, src in enumerate(to_move):
            if src != dest_index:
                # we need to decrement the src_i by 1 for each time we have
                # previously pulled items out from in front of the src_i
                src -= sum(x <= src for x in popped)
                # if source is past the insertion point, increment src for each
                # previous insertion
                if src >= dest_index:
                    src += i
                yield src, dest_index + d_inc

            popped.append(src)
            # if the item moved up, icrement the destination index
            if dest_index <= src:
                d_inc += 1

    def _connect_child_emitters(self, child: _T) -> None:
        """Connect all events from the child to be reemitted."""
        for emitter in iter_signal_instances(child):
            emitter.connect(self._reemit_child_event)

    def _disconnect_child_emitters(self, child: _T) -> None:
        """Disconnect all events from the child from the reemitter."""
        for emitter in iter_signal_instances(child):
            emitter.disconnect(self._reemit_child_event)

    def _reemit_child_event(self, *args: Any) -> None:
        """Re-emit event from child with index."""
        emitter = Signal.current_emitter()
        if emitter is None:
            return  # pragma: no cover
        obj = emitter.instance
        try:
            idx = self.index(obj)
        except ValueError:  # pragma: no cover
            return

        if (
            args
            and isinstance(emitter, SignalRelay)
            and isinstance(args[0], EmissionInfo)
        ):
            emitter, args = args[0]

        self.events.child_event.emit(idx, obj, emitter, args)

    # PYDANTIC SUPPORT

    @classmethod
    def __get_pydantic_core_schema__(
        cls, source_type: Any, handler: Callable
    ) -> Mapping[str, Any]:
        """Return the Pydantic core schema for this object."""
        from pydantic_core import core_schema

        args = get_args(source_type)
        return core_schema.no_info_after_validator_function(
            function=cls,
            schema=core_schema.list_schema(
                items_schema=handler(args[0]) if args else None,
            ),
        )

copy() #

Return a shallow copy of the list.

Source code in psygnal/containers/_evented_list.py
219
220
221
def copy(self) -> Self:
    """Return a shallow copy of the list."""
    return self.__newlike__(self)

insert(index, value) #

Insert value before index.

Source code in psygnal/containers/_evented_list.py
139
140
141
142
143
144
145
def insert(self, index: int, value: _T) -> None:
    """Insert `value` before index."""
    _value = self._pre_insert(value)
    self.events.inserting.emit(index)
    self._data.insert(index, _value)
    self.events.inserted.emit(index, value)
    self._post_insert(value)

move(src_index, dest_index=0) #

Insert object at src_index before dest_index.

Both indices refer to the list prior to any object removal (pre-move space).

Source code in psygnal/containers/_evented_list.py
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
def move(self, src_index: int, dest_index: int = 0) -> bool:
    """Insert object at `src_index` before `dest_index`.

    Both indices refer to the list prior to any object removal
    (pre-move space).
    """
    if dest_index < 0:
        dest_index += len(self) + 1
    if dest_index in (src_index, src_index + 1):
        # this is a no-op
        return False

    self.events.moving.emit(src_index, dest_index)
    item = self._data.pop(src_index)
    if dest_index > src_index:
        dest_index -= 1
    self._data.insert(dest_index, item)
    self.events.moved.emit(src_index, dest_index, item)
    self.events.reordered.emit()
    return True

move_multiple(sources, dest_index=0) #

Move a batch of sources indices, to a single destination.

Note, if dest_index is higher than any of the sources, then the resulting position of the moved objects after the move operation is complete will be lower than dest_index.

Parameters:

  • sources (Iterable[Union[int, slice]]) –

    A sequence of indices

  • dest_index (int, optional) –

    The destination index. All sources will be inserted before this index (in pre-move space), by default 0... which has the effect of "bringing to front" everything in sources, or acting as a "reorder" method if sources contains all indices.

Returns:

  • int

    The number of successful move operations completed.

Raises:

  • TypeError

    If the destination index is a slice, or any of the source indices are not int or slice.

Source code in psygnal/containers/_evented_list.py
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
def move_multiple(self, sources: Iterable[Index], dest_index: int = 0) -> int:
    """Move a batch of `sources` indices, to a single destination.

    Note, if `dest_index` is higher than any of the `sources`, then
    the resulting position of the moved objects after the move operation
    is complete will be lower than `dest_index`.

    Parameters
    ----------
    sources : Iterable[Union[int, slice]]
        A sequence of indices
    dest_index : int, optional
        The destination index.  All sources will be inserted before this
        index (in pre-move space), by default 0... which has the effect of
        "bringing to front" everything in `sources`, or acting as a
        "reorder" method if `sources` contains all indices.

    Returns
    -------
    int
        The number of successful move operations completed.

    Raises
    ------
    TypeError
        If the destination index is a slice, or any of the source indices
        are not `int` or `slice`.
    """
    # calling list here makes sure that there are no index errors up front
    move_plan = list(self._move_plan(sources, dest_index))

    # don't assume index adjacency ... so move objects one at a time
    # this *could* be simplified with an intermediate list ... but this way
    # allows any views (such as QtViews) to update themselves more easily.
    # If this needs to be changed in the future for performance reasons,
    # then the associated QtListView will need to changed from using
    # `beginMoveRows` & `endMoveRows` to using `layoutAboutToBeChanged` &
    # `layoutChanged` while *manually* updating model indices with
    # `changePersistentIndexList`.  That becomes much harder to do with
    # nested tree-like models.
    with self.events.reordered.blocked():
        for src, dest in move_plan:
            self.move(src, dest)

    self.events.reordered.emit()
    return len(move_plan)

reverse(*, emit_individual_events=False) #

Reverse list IN PLACE.

Source code in psygnal/containers/_evented_list.py
265
266
267
268
269
270
271
def reverse(self, *, emit_individual_events: bool = False) -> None:
    """Reverse list *IN PLACE*."""
    if emit_individual_events:
        super().reverse()
    else:
        self._data.reverse()
    self.events.reordered.emit()

psygnal.containers.EventedSet #

Bases: _BaseMutableSet[_T]

A set with an items_changed signal that emits when items are added/removed.

Parameters:

  • iterable (Iterable[_T]) –

    Data to populate the set. If omitted, an empty set is created.

Attributes:

  • events (SetEvents) –

    SignalGroup that with events related to set mutation. (see SetEvents)

Examples:

>>> from psygnal.containers import EventedSet
>>>
>>> my_set = EventedSet([1, 2, 3])
>>> my_set.events.items_changed.connect(
>>>     lambda a, r: print(f"added={a}, removed={r}")
>>> )
>>> my_set.update({3, 4, 5})
added=(4, 5), removed=()

Multi-item events will be reduced into a single emission:

>>> my_set.symmetric_difference_update({4, 5, 6, 7})
added=(6, 7), removed=(4, 5)
>>> my_set
EventedSet({1, 2, 3, 6, 7})
Source code in psygnal/containers/_evented_set.py
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
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
class EventedSet(_BaseMutableSet[_T]):
    """A set with an `items_changed` signal that emits when items are added/removed.

    Parameters
    ----------
    iterable : Iterable[_T]
        Data to populate the set.  If omitted, an empty set is created.

    Attributes
    ----------
    events : SetEvents
        SignalGroup that with events related to set mutation.  (see SetEvents)

    Examples
    --------
    >>> from psygnal.containers import EventedSet
    >>>
    >>> my_set = EventedSet([1, 2, 3])
    >>> my_set.events.items_changed.connect(
    >>>     lambda a, r: print(f"added={a}, removed={r}")
    >>> )
    >>> my_set.update({3, 4, 5})
    added=(4, 5), removed=()

    Multi-item events will be reduced into a single emission:
    >>> my_set.symmetric_difference_update({4, 5, 6, 7})
    added=(6, 7), removed=(4, 5)

    >>> my_set
    EventedSet({1, 2, 3, 6, 7})
    """

    events: SetEvents  # pragma: no cover

    def __init__(self, iterable: Iterable[_T] = ()):
        self.events = self._get_events_class()
        super().__init__(iterable)

    def update(self, *others: Iterable[_T]) -> None:
        """Update this set with the union of this set and others."""
        with self.events.items_changed.paused(_reduce_events):
            super().update(*others)

    def clear(self) -> None:
        """Remove all elements from this set."""
        with self.events.items_changed.paused(_reduce_events):
            super().clear()

    def difference_update(self, *s: Iterable[_T]) -> None:
        """Remove all elements of another set from this set."""
        with self.events.items_changed.paused(_reduce_events):
            super().difference_update(*s)

    def intersection_update(self, *s: Iterable[_T]) -> None:
        """Update this set with the intersection of itself and another."""
        with self.events.items_changed.paused(_reduce_events):
            super().intersection_update(*s)

    def symmetric_difference_update(self, __s: Iterable[_T]) -> None:
        """Update this set with the symmetric difference of itself and another.

        This will remove any items in this set that are also in `other`, and
        add any items in others that are not present in this set.
        """
        with self.events.items_changed.paused(_reduce_events, ((), ())):
            super().symmetric_difference_update(__s)

    def _pre_add_hook(self, item: _T) -> _T | BailType:
        return BAIL if item in self else item

    def _post_add_hook(self, item: _T) -> None:
        self._emit_change((item,), ())

    def _pre_discard_hook(self, item: _T) -> _T | BailType:
        return BAIL if item not in self else item

    def _post_discard_hook(self, item: _T) -> None:
        self._emit_change((), (item,))

    def _emit_change(self, added: tuple[_T, ...], removed: tuple[_T, ...]) -> None:
        """Emit a change event."""
        self.events.items_changed.emit(added, removed)

    def _get_events_class(self) -> SetEvents:
        return SetEvents()

clear() #

Remove all elements from this set.

Source code in psygnal/containers/_evented_set.py
267
268
269
270
def clear(self) -> None:
    """Remove all elements from this set."""
    with self.events.items_changed.paused(_reduce_events):
        super().clear()

difference_update(*s) #

Remove all elements of another set from this set.

Source code in psygnal/containers/_evented_set.py
272
273
274
275
def difference_update(self, *s: Iterable[_T]) -> None:
    """Remove all elements of another set from this set."""
    with self.events.items_changed.paused(_reduce_events):
        super().difference_update(*s)

intersection_update(*s) #

Update this set with the intersection of itself and another.

Source code in psygnal/containers/_evented_set.py
277
278
279
280
def intersection_update(self, *s: Iterable[_T]) -> None:
    """Update this set with the intersection of itself and another."""
    with self.events.items_changed.paused(_reduce_events):
        super().intersection_update(*s)

symmetric_difference_update(__s) #

Update this set with the symmetric difference of itself and another.

This will remove any items in this set that are also in other, and add any items in others that are not present in this set.

Source code in psygnal/containers/_evented_set.py
282
283
284
285
286
287
288
289
def symmetric_difference_update(self, __s: Iterable[_T]) -> None:
    """Update this set with the symmetric difference of itself and another.

    This will remove any items in this set that are also in `other`, and
    add any items in others that are not present in this set.
    """
    with self.events.items_changed.paused(_reduce_events, ((), ())):
        super().symmetric_difference_update(__s)

update(*others) #

Update this set with the union of this set and others.

Source code in psygnal/containers/_evented_set.py
262
263
264
265
def update(self, *others: Iterable[_T]) -> None:
    """Update this set with the union of this set and others."""
    with self.events.items_changed.paused(_reduce_events):
        super().update(*others)

psygnal.containers.EventedOrderedSet #

Bases: EventedSet, OrderedSet[_T]

A ordered variant of EventedSet that maintains insertion order.

Parameters:

  • iterable (Iterable[_T]) –

    Data to populate the set. If omitted, an empty set is created.

Attributes:

  • events (SetEvents) –

    SignalGroup that with events related to set mutation. (see SetEvents)

Source code in psygnal/containers/_evented_set.py
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
class EventedOrderedSet(EventedSet, OrderedSet[_T]):
    """A ordered variant of EventedSet that maintains insertion order.

    Parameters
    ----------
    iterable : Iterable[_T]
        Data to populate the set.  If omitted, an empty set is created.

    Attributes
    ----------
    events : SetEvents
        SignalGroup that with events related to set mutation.  (see SetEvents)
    """

    # reproducing init here to avoid a mkdocs warning:
    # "Parameter 'iterable' does not appear in the function signature"
    def __init__(self, iterable: Iterable[_T] = ()):
        super().__init__(iterable)

psygnal.containers.Selection #

Bases: EventedOrderedSet[_T]

An model of selected items, with a active and current item.

There can only be one active and one current item, but there can be multiple selected items. An "active" item is defined as a single selected item (if multiple items are selected, there is no active item). The "current" item is mostly useful for (e.g.) keyboard actions: even with multiple items selected, you may only have one current item, and keyboard events (like up and down) can modify that current item. It's possible to have a current item without an active item, but an active item will always be the current item.

An item can be the current item and selected at the same time. Qt views will ensure that there is always a current item as keyboard navigation, for example, requires a current item. This pattern mimics current/selected items from Qt: https://doc.qt.io/qt-5/model-view-programming.html#current-item-and-selected-items

Parameters:

  • data (iterable, optional) –

    Elements to initialize the set with.

  • parent (Container, optional) –

    The parent container, if any. This is used to provide validation upon mutation in common use cases.

Attributes:

  • events (SelectionEvents) –

    SignalGroup that with events related to selection changes. (see SelectionEvents)

  • active (Any, optional) –

    The active item, if any. An "active" item is defined as a single selected item (if multiple items are selected, there is no active item)

  • _current (Any, optional) –

    The current item, if any. This is used primarily by GUI views when handling mouse/key events.

Source code in psygnal/containers/_selection.py
 37
 38
 39
 40
 41
 42
 43
 44
 45
 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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class Selection(EventedOrderedSet[_T]):
    """An model of selected items, with a `active` and `current` item.

    There can only be one `active` and one `current` item, but there can be
    multiple selected items.  An "active" item is defined as a single selected
    item (if multiple items are selected, there is no active item).  The
    "current" item is mostly useful for (e.g.) keyboard actions: even with
    multiple items selected, you may only have one current item, and keyboard
    events (like up and down) can modify that current item.  It's possible to
    have a current item without an active item, but an active item will always
    be the current item.

    An item can be the current item and selected at the same time. Qt views
    will ensure that there is always a current item as keyboard navigation,
    for example, requires a current item.
    This pattern mimics current/selected items from Qt:
    https://doc.qt.io/qt-5/model-view-programming.html#current-item-and-selected-items

    Parameters
    ----------
    data : iterable, optional
        Elements to initialize the set with.
    parent : Container, optional
        The parent container, if any. This is used to provide validation upon
        mutation in common use cases.

    Attributes
    ----------
    events : SelectionEvents
        SignalGroup that with events related to selection changes. (see SelectionEvents)
    active : Any, optional
        The active item, if any. An "active" item is defined as a single selected
        item (if multiple items are selected, there is no active item)
    _current : Any, optional
        The current item, if any. This is used primarily by GUI views when
        handling mouse/key events.
    """

    events: SelectionEvents  # pragma: no cover

    def __init__(self, data: Iterable[_T] = (), parent: Container | None = None):
        self._active: _T | None = None
        self._current_: _T | None = None
        self._parent: Container | None = parent
        super().__init__(iterable=data)
        self._update_active()

    @property
    def _current(self) -> _T | None:  # pragma: no cover
        """Get current item."""
        return self._current_

    @_current.setter
    def _current(self, value: _T | None) -> None:  # pragma: no cover
        """Set current item."""
        if value == self._current_:
            return
        self._current_ = value
        self.events._current.emit(value)

    @property
    def active(self) -> _T | None:  # pragma: no cover
        """Return the currently active item or None."""
        return self._active

    @active.setter
    def active(self, value: _T | None) -> None:  # pragma: no cover
        """Set the active item.

        This makes `value` the only selected item, and makes it current.
        """
        if value == self._active:
            return
        self._active = value
        self.clear() if value is None else self.select_only(value)
        self._current = value
        self.events.active.emit(value)

    def clear(self, keep_current: bool = False) -> None:
        """Clear the selection.

        Parameters
        ----------
        keep_current : bool
            If `False` (the default), the "current" item will also be set to None.
        """
        if not keep_current:
            self._current = None
        super().clear()

    def toggle(self, obj: _T) -> None:
        """Toggle selection state of obj."""
        self.symmetric_difference_update({obj})

    def select_only(self, obj: _T) -> None:
        """Unselect everything but `obj`. Add to selection if not currently selected."""
        self.intersection_update({obj})
        self.add(obj)

    def _update_active(self) -> None:
        """On a selection event, update the active item based on selection.

        An active item is a single selected item.
        """
        if len(self) == 1:
            self.active = next(iter(self))
        elif self._active is not None:
            self._active = None
            self.events.active.emit(None)

    def _get_events_class(self) -> SelectionEvents:
        """Override SetEvents with SelectionEvents."""
        return SelectionEvents()

    def _emit_change(self, added: tuple[_T, ...], removed: tuple[_T, ...]) -> None:
        """Emit a change event."""
        super()._emit_change(added, removed)
        self._update_active()

    def _pre_add_hook(self, item: _T) -> _T | BailType:
        if self._parent is not None and item not in self._parent:
            raise ValueError(
                "Cannot select an item that is not in the parent container."
            )
        return super()._pre_add_hook(item)

    def __hash__(self) -> int:
        """Make selection hashable."""
        return id(self)

active: _T | None property writable #

Return the currently active item or None.

clear(keep_current=False) #

Clear the selection.

Parameters:

  • keep_current (bool) –

    If False (the default), the "current" item will also be set to None.

Source code in psygnal/containers/_selection.py
115
116
117
118
119
120
121
122
123
124
125
def clear(self, keep_current: bool = False) -> None:
    """Clear the selection.

    Parameters
    ----------
    keep_current : bool
        If `False` (the default), the "current" item will also be set to None.
    """
    if not keep_current:
        self._current = None
    super().clear()

select_only(obj) #

Unselect everything but obj. Add to selection if not currently selected.

Source code in psygnal/containers/_selection.py
131
132
133
134
def select_only(self, obj: _T) -> None:
    """Unselect everything but `obj`. Add to selection if not currently selected."""
    self.intersection_update({obj})
    self.add(obj)

toggle(obj) #

Toggle selection state of obj.

Source code in psygnal/containers/_selection.py
127
128
129
def toggle(self, obj: _T) -> None:
    """Toggle selection state of obj."""
    self.symmetric_difference_update({obj})

psygnal.containers.SelectableEventedList #

Bases: Selectable[_T], EventedList[_T]

EventedList subclass with a built in selection model.

In addition to all EventedList properties, this class also has a selection attribute that manages a set of selected items in the list.

Parameters:

  • data (iterable, optional) –

    Elements to initialize the list with.

  • hashable (bool) –

    Whether the list should be hashable as id(self). By default True.

  • child_events (bool) –

    Whether to re-emit events from emitted from evented items in the list (i.e. items that have SignalInstances). If True, child events can be connected at EventedList.events.child_event. By default, False.

Attributes:

  • events (ListEvents) –

    SignalGroup that with events related to list mutation. (see ListEvents)

  • selection (Selection) –

    An evented set containing the currently selected items, along with an active and current item. (See Selection)

Source code in psygnal/containers/_selectable_evented_list.py
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 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
class SelectableEventedList(Selectable[_T], EventedList[_T]):
    """`EventedList` subclass with a built in selection model.

    In addition to all `EventedList` properties, this class also has a `selection`
    attribute that manages a set of selected items in the list.

    Parameters
    ----------
    data : iterable, optional
        Elements to initialize the list with.
    hashable : bool
        Whether the list should be hashable as id(self). By default `True`.
    child_events: bool
        Whether to re-emit events from emitted from evented items in the list
        (i.e. items that have SignalInstances). If `True`, child events can be connected
        at `EventedList.events.child_event`. By default, `False`.

    Attributes
    ----------
    events : ListEvents
        SignalGroup that with events related to list mutation.  (see ListEvents)
    selection : Selection
        An evented set containing the currently selected items, along with an `active`
        and `current` item.  (See `Selection`)
    """

    events: ListEvents  # pragma: no cover

    def __init__(
        self,
        data: Iterable[_T] = (),
        *,
        hashable: bool = True,
        child_events: bool = False,
    ):
        self._activate_on_insert: bool = True
        super().__init__(data=data, hashable=hashable, child_events=child_events)
        self.events.removed.connect(self._on_item_removed)

    def _on_item_removed(self, idx: int, obj: Any) -> None:
        self.selection.discard(obj)

    def insert(self, index: int, value: _T) -> None:
        """Insert item(s) into the list and update the selection."""
        super().insert(index, value)
        if self._activate_on_insert:
            self.selection.active = value

    def select_all(self) -> None:
        """Select all items in the list."""
        self.selection.update(self)

    def deselect_all(self) -> None:
        """Deselect all items in the list."""
        self.selection.clear()

    def select_next(
        self, step: int = 1, expand_selection: bool = False, wraparound: bool = False
    ) -> None:
        """Select the next item in the list.

        Parameters
        ----------
        step : int
            The step size to take when picking the next item, by default 1
        expand_selection : bool
            If True, will expand the selection to contain the both the current item and
            the next item, by default False
        wraparound : bool
            Whether to return to the beginning of the list of the end has been reached,
            by default False
        """
        if len(self) == 0:
            return
        elif not self.selection:
            idx = -1 if step > 0 else 0
        else:
            idx = self.index(self.selection._current) + step
        idx_in_sequence = len(self) > idx >= 0
        if wraparound:
            idx = idx % len(self)
        elif not idx_in_sequence:
            idx = -1 if step > 0 else 0
        next_item = self[idx]
        if expand_selection:
            self.selection.add(next_item)
            self.selection._current = next_item
        else:
            self.selection.active = next_item

    def select_previous(
        self, expand_selection: bool = False, wraparound: bool = False
    ) -> None:
        """Select the previous item in the list."""
        self.select_next(
            step=-1, expand_selection=expand_selection, wraparound=wraparound
        )

    def remove_selected(self) -> Tuple[_T, ...]:
        """Remove selected items from the list and the selection.

        Returns
        -------
        Tuple[_T, ...]
            The items that were removed.
        """
        selected_items = tuple(self.selection)
        idx = 0
        for item in list(self.selection):
            idx = self.index(item)
            self.remove(item)
        new_idx = max(0, idx - 1)
        if len(self) > new_idx:
            self.selection.add(self[new_idx])
        return selected_items

deselect_all() #

Deselect all items in the list.

Source code in psygnal/containers/_selectable_evented_list.py
63
64
65
def deselect_all(self) -> None:
    """Deselect all items in the list."""
    self.selection.clear()

insert(index, value) #

Insert item(s) into the list and update the selection.

Source code in psygnal/containers/_selectable_evented_list.py
53
54
55
56
57
def insert(self, index: int, value: _T) -> None:
    """Insert item(s) into the list and update the selection."""
    super().insert(index, value)
    if self._activate_on_insert:
        self.selection.active = value

remove_selected() #

Remove selected items from the list and the selection.

Returns:

  • Tuple[_T, ...]

    The items that were removed.

Source code in psygnal/containers/_selectable_evented_list.py
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def remove_selected(self) -> Tuple[_T, ...]:
    """Remove selected items from the list and the selection.

    Returns
    -------
    Tuple[_T, ...]
        The items that were removed.
    """
    selected_items = tuple(self.selection)
    idx = 0
    for item in list(self.selection):
        idx = self.index(item)
        self.remove(item)
    new_idx = max(0, idx - 1)
    if len(self) > new_idx:
        self.selection.add(self[new_idx])
    return selected_items

select_all() #

Select all items in the list.

Source code in psygnal/containers/_selectable_evented_list.py
59
60
61
def select_all(self) -> None:
    """Select all items in the list."""
    self.selection.update(self)

select_next(step=1, expand_selection=False, wraparound=False) #

Select the next item in the list.

Parameters:

  • step (int) –

    The step size to take when picking the next item, by default 1

  • expand_selection (bool) –

    If True, will expand the selection to contain the both the current item and the next item, by default False

  • wraparound (bool) –

    Whether to return to the beginning of the list of the end has been reached, by default False

Source code in psygnal/containers/_selectable_evented_list.py
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
def select_next(
    self, step: int = 1, expand_selection: bool = False, wraparound: bool = False
) -> None:
    """Select the next item in the list.

    Parameters
    ----------
    step : int
        The step size to take when picking the next item, by default 1
    expand_selection : bool
        If True, will expand the selection to contain the both the current item and
        the next item, by default False
    wraparound : bool
        Whether to return to the beginning of the list of the end has been reached,
        by default False
    """
    if len(self) == 0:
        return
    elif not self.selection:
        idx = -1 if step > 0 else 0
    else:
        idx = self.index(self.selection._current) + step
    idx_in_sequence = len(self) > idx >= 0
    if wraparound:
        idx = idx % len(self)
    elif not idx_in_sequence:
        idx = -1 if step > 0 else 0
    next_item = self[idx]
    if expand_selection:
        self.selection.add(next_item)
        self.selection._current = next_item
    else:
        self.selection.active = next_item

select_previous(expand_selection=False, wraparound=False) #

Select the previous item in the list.

Source code in psygnal/containers/_selectable_evented_list.py
101
102
103
104
105
106
107
def select_previous(
    self, expand_selection: bool = False, wraparound: bool = False
) -> None:
    """Select the previous item in the list."""
    self.select_next(
        step=-1, expand_selection=expand_selection, wraparound=wraparound
    )

Container SignalGroups#

psygnal.containers._evented_dict.DictEvents #

Bases: SignalGroup

Events available on EventedDict.

Attributes:

  • adding (Signal[Any]) –

    (key,) emitted before an item is added at key

  • added (Signal[Any, Any]) –

    (key, value) emitted after a value is added at key

  • changing (Signal[Any, Any, Any]) –

    (key, old_value, new_value) emitted before old_value is replaced with new_value at key

  • changed (Signal[Any, Any, Any]) –

    (key, old_value, new_value) emitted before old_value is replaced with new_value at key

  • removing (Signal[Any]) –

    (key,) emitted before an item is removed at key

  • removed (Signal[Any, Any]) –

    (key, value) emitted after value is removed at index

psygnal.containers._evented_list.ListEvents #

Bases: SignalGroup

Events available on EventedList.

Attributes:

  • inserting (Signal[int]) –

    (index) emitted before an item is inserted at index

  • inserted (Signal[int, Any]) –

    (index, value) emitted after value is inserted at index

  • removing (Signal[int]) –

    (index) emitted before an item is removed at index

  • removed (Signal[int, Any]) –

    (index, value) emitted after value is removed at index

  • moving (Signal[int, int]) –

    (index, new_index) emitted before an item is moved from index to new_index

  • moved (Signal[int, int, Any]) –

    (index, new_index, value) emitted after value is moved from index to new_index

  • changed (Signal[Union[int, slice], Any, Any]) –

    (index_or_slice, old_value, value) emitted when index is set from old_value to value

  • reordered (Signal) –

    emitted when the list is reordered (eg. moved/reversed).

  • child_event (Signal[int, Any, SignalInstance, tuple]) –

    (index, object, emitter, args) emitted when an object in the list emits an event. Note that the EventedList must be created with child_events=True in order for this to be emitted.

psygnal.containers._evented_set.SetEvents #

Bases: SignalGroup

Events available on EventedSet.

Attributes:

  • items_changed (added (Tuple[Any, ...], removed: Tuple[Any, ...])) –

    A signal that will emitted whenever an item or items are added or removed. Connected callbacks will be called with callback(added, removed), where added and removed are tuples containing the objects that have been added or removed from the set.

psygnal.containers._selection.SelectionEvents #

Bases: SetEvents

Events available on Selection.

Attributes:

  • items_changed (added (Tuple[_T], removed: Tuple[_T])) –

    A signal that will emitted whenever an item or items are added or removed. Connected callbacks will be called with callback(added, removed), where added and removed are tuples containing the objects that have been added or removed from the set.

  • active (value (_T)) –

    Emitted when the active item has changed. An active item is a single selected item.

  • _current (value (_T)) –

    Emitted when the current item has changed. (Private event)