mode.utils.collections

Custom data structures.

AttributeDict

Bases: dict, AttributeDictMixin

Dict subclass with attribute access.

Source code in mode/utils/collections.py

class AttributeDict(dict, AttributeDictMixin):
    """Dict subclass with attribute access."""

AttributeDictMixin

Mixin for Mapping interface that adds attribute access.

Example:

d.key -> d[key]

Source code in mode/utils/collections.py

class AttributeDictMixin:
    """Mixin for Mapping interface that adds attribute access.

    Example:

    `d.key` -> `d[key]`
    """

    def __getattr__(self, k: str) -> Any:
        """`d.key -> d[key]`."""
        try:
            return cast(Mapping, self)[k]
        except KeyError as err:
            raise AttributeError(
                f"{type(self).__name__!r} object has no attribute {k!r}"
            ) from err

    def __setattr__(self, key: str, value: Any) -> None:
        """
        ```python
        d[key] = value -> d.key = value
        ```
        """
        self[key] = value

__getattr__(k)

d.key -> d[key].

Source code in mode/utils/collections.py

def __getattr__(self, k: str) -> Any:
    """`d.key -> d[key]`."""
    try:
        return cast(Mapping, self)[k]
    except KeyError as err:
        raise AttributeError(
            f"{type(self).__name__!r} object has no attribute {k!r}"
        ) from err

__setattr__(key, value)

d[key] = value -> d.key = value

Source code in mode/utils/collections.py

def __setattr__(self, key: str, value: Any) -> None:
    """
    ```python
    d[key] = value -> d.key = value
    ```
    """
    self[key] = value

DictAttribute

Bases: MutableMapping[str, VT], MappingViewProxy

Dict interface to attributes.

• obj[k] -> obj.k
• obj[k] = val -> obj.k = val

Source code in mode/utils/collections.py

class DictAttribute(MutableMapping[str, VT], MappingViewProxy):
    """Dict interface to attributes.

    - `obj[k]` -> `obj.k`
    - `obj[k] = val` -> `obj.k = val`
    """

    obj: Any = None

    def __init__(self, obj: Any) -> None:
        object.__setattr__(self, "obj", obj)

    def __getattr__(self, key: Any) -> Any:
        return getattr(self.obj, key)

    def __setattr__(self, key: Any, value: Any) -> None:
        return setattr(self.obj, key, value)

    def get(self, key: Any, default: Any = None) -> Any:
        try:
            return self[key]
        except KeyError:
            return default

    def setdefault(self, key: Any, default: Any = None) -> Any:
        if key in self:
            return self[key]
        self[key] = default
        return default

    def __getitem__(self, key: Any) -> Any:
        try:
            return getattr(self.obj, key)
        except AttributeError as err:
            raise KeyError(key) from err

    def __setitem__(self, key: Any, value: Any) -> None:
        setattr(self.obj, key, value)

    def __delitem__(self, key: Any) -> None:
        raise NotImplementedError()

    def __len__(self) -> int:
        return len(self.obj.__dict__)

    def __contains__(self, key: Any) -> bool:
        return hasattr(self.obj, key)

    def __iter__(self) -> Iterator[str]:
        return self._keys()

    def _keys(self) -> Iterator[str]:
        yield from dir(self.obj)

    def _values(self) -> Iterator[str]:
        obj = self.obj
        for key in self:
            yield getattr(obj, key)

    def _items(self) -> Iterator[tuple[str, VT]]:
        obj = self.obj
        for key in self:
            yield key, getattr(obj, key)

FastUserDict

Bases: MutableMapping[KT, VT]

Proxy to dict.

Like collection.UserDict but reimplements some methods for better performance when the underlying dictionary is a real dict.

Source code in mode/utils/collections.py

class FastUserDict(MutableMapping[KT, VT]):
    """Proxy to dict.

    Like `collection.UserDict` but reimplements some methods
    for better performance when the underlying dictionary is a real dict.
    """

    data: MutableMapping[KT, VT]

    @classmethod
    def fromkeys(
        cls, iterable: Iterable[KT], value: VT = None
    ) -> "FastUserDict":
        d = cls()
        d.update(dict.fromkeys(iterable, value))
        return d

    def __getitem__(self, key: KT) -> VT:
        if not hasattr(self, "__missing__"):
            return self.data[key]
        if key in self.data:
            return self.data[key]
        return self.__missing__(key)  # type: ignore

    def __setitem__(self, key: KT, value: VT) -> None:
        self.data[key] = value

    def __delitem__(self, key: KT) -> None:
        del self.data[key]

    def __len__(self) -> int:
        return len(self.data)

    def __iter__(self) -> Iterator[KT]:
        return iter(self.data)

    # Rest is fast versions of generic slow MutableMapping methods.

    def __contains__(self, key: object) -> bool:
        return key in self.data

    def __repr__(self) -> str:
        return repr(self.data)

    def copy(self) -> dict:
        return dict(self.data)

    def update(self, *args: Any, **kwargs: Any) -> None:
        self.data.update(*args, **kwargs)

    def clear(self) -> None:
        self.data.clear()

    def items(self) -> ItemsView:
        return cast(ItemsView, self.data.items())

    def keys(self) -> KeysView:
        return cast(KeysView, self.data.keys())

    def values(self) -> ValuesView:
        return self.data.values()

FastUserList

Bases: UserList

Proxy to list.

Source code in mode/utils/collections.py

class FastUserList(UserList):
    """Proxy to list."""

FastUserSet

Bases: MutableSet[T]

Proxy to set.

Source code in mode/utils/collections.py

class FastUserSet(MutableSet[T]):
    """Proxy to set."""

    data: MutableSet[T]

    # -- Immutable Methods --

    def __and__(self, other: Set[Any]) -> MutableSet[T]:
        return cast(MutableSet, self.data.__and__(other))

    def __contains__(self, key: Any) -> bool:
        return self.data.__contains__(key)

    def __ge__(self, other: Set[T]) -> bool:
        return self.data.__ge__(other)

    def __iter__(self) -> Iterator[T]:
        return iter(self.data)

    def __le__(self, other: Set[T]) -> bool:
        return self.data.__le__(other)

    def __len__(self) -> int:
        return len(self.data)

    def __or__(self, other: Set[_S]) -> Set[Union[T, _S]]:
        return self.data.__or__(other)

    def __rand__(self, other: Set[T]) -> MutableSet[T]:
        return self.__and__(other)

    def __reduce__(self) -> tuple:
        return self.data.__reduce__()  # type: ignore

    def __reduce_ex__(self, protocol: Any) -> tuple:
        return self.data.__reduce_ex__(protocol)  # type: ignore

    def __repr__(self) -> str:
        return repr(self.data)

    def __ror__(self, other: Set[T]) -> MutableSet[T]:
        return cast(MutableSet, self.data.__ror__(other))  # type: ignore

    def __rsub__(self, other: Set[T]) -> MutableSet[T]:
        return cast(MutableSet, other.__rsub__(self.data))  # type: ignore

    def __rxor__(self, other: Set[T]) -> MutableSet[T]:
        return cast(MutableSet, other.__rxor__(self.data))  # type: ignore

    def __sizeof__(self) -> int:
        return self.data.__sizeof__()

    def __str__(self) -> str:
        return str(self.data)

    def __sub__(self, other: Set[Any]) -> MutableSet[T]:
        return cast(MutableSet[T], self.data.__sub__(other))

    def __xor__(self, other: Set) -> MutableSet[T]:
        return cast(MutableSet, self.data.__xor__(other))

    def copy(self) -> MutableSet[T]:
        return set(self.data)

    def difference(self, other: _Setlike[T]) -> MutableSet[T]:
        return self.data.difference(other)  # type: ignore

    def intersection(self, other: _Setlike[T]) -> MutableSet[T]:
        return self.data.intersection(other)  # type: ignore

    def isdisjoint(self, other: Iterable[T]) -> bool:
        return self.data.isdisjoint(other)

    def issubset(self, other: Set[T]) -> bool:
        return self.data.issubset(other)  # type: ignore

    def issuperset(self, other: Set[T]) -> bool:
        return self.data.issuperset(other)  # type: ignore

    def symmetric_difference(self, other: _Setlike[T]) -> MutableSet[T]:
        return cast(MutableSet, self.data.symmetric_difference(other))  # type: ignore

    def union(self, other: _Setlike[T]) -> MutableSet[T]:
        return cast(MutableSet, self.data.union(other))  # type: ignore

    # -- Mutable Methods --

    def __iand__(self, other: Set[Any]) -> "FastUserSet":
        self.data.__iand__(other)
        return self

    def __ior__(self, other: Set[_S]) -> "FastUserSet":
        cast(MutableSet[Union[T, _S]], self.data).__ior__(other)
        return self

    def __isub__(self, other: Set[Any]) -> "FastUserSet[T]":
        self.data.__isub__(other)
        return self

    def __ixor__(self, other: Set[_S]) -> "FastUserSet":
        cast(MutableSet[Union[T, _S]], self.data).__ixor__(other)
        return self

    def add(self, value: T) -> None:
        self.data.add(value)

    def clear(self) -> None:
        self.data.clear()

    def difference_update(self, other: _Setlike[T]) -> None:
        self.data.difference_update(other)  # type: ignore

    def discard(self, value: T) -> None:
        self.data.discard(value)

    def intersection_update(self, other: _Setlike[T]) -> None:
        self.data.intersection_update(other)  # type: ignore

    def pop(self) -> T:
        return self.data.pop()

    def remove(self, value: T) -> None:
        self.data.remove(value)

    def symmetric_difference_update(self, other: _Setlike[T]) -> None:
        self.data.symmetric_difference_update(other)  # type: ignore

    def update(self, other: _Setlike[T]) -> None:
        self.data.update(other)  # type: ignore

Heap

Bases: MutableSequence[_ComparableT]

Generic interface to heapq.

Elements have to be orderable: heapq maintains the heap invariant by comparing them, so the element type is bound to _typeshed.SupportsRichComparison -- anything defining __lt__ or __gt__. Heap[int] and Heap[str] are fine, while an element type with no ordering is rejected by the type checker instead of failing at the first push.

Source code in mode/utils/collections.py

class Heap(MutableSequence[_ComparableT]):
    """Generic interface to `heapq`.

    Elements have to be orderable: `heapq` maintains the heap invariant by
    comparing them, so the element type is bound to
    `_typeshed.SupportsRichComparison` -- anything defining `__lt__` or
    `__gt__`.  `Heap[int]` and `Heap[str]` are fine, while an element type
    with no ordering is rejected by the type checker instead of failing at
    the first `push`.
    """

    def __init__(self, data: Optional[Sequence[_ComparableT]] = None) -> None:
        self.data = list(data or [])
        heapify(self.data)

    def pop(self, index: int = 0) -> _ComparableT:
        """Pop the smallest item off the heap.

        Maintains the heap invariant.
        """
        if index == 0:
            return heappop(self.data)
        else:
            raise NotImplementedError(
                "Heap can only pop index 0, please use h.data.pop(index)"
            )

    def push(self, item: _ComparableT) -> None:
        """Push item onto heap, maintaining the heap invariant."""
        heappush(self.data, item)

    def pushpop(self, item: _ComparableT) -> _ComparableT:
        """Push item on the heap, then pop and return from the heap.

        The combined action runs more efficiently than
        `push` followed by a separate call to `pop`.
        """
        return heappushpop(self.data, item)

    def replace(self, item: _ComparableT) -> _ComparableT:
        """Pop and return the current smallest value, and add the new item.

        This is more efficient than :meth`pop` followed by `push`,
        and can be more appropriate when using a fixed-size heap.

        Note that the value returned may be larger than item!
        That constrains reasonable uses of this routine unless written as
        part of a conditional replacement:

        ```python
        if item > heap[0]:
            item = heap.replace(item)
        ```
        """
        return heapreplace(self.data, item)

    def nlargest(
        self, n: int, key: Optional[Callable] = None
    ) -> list[_ComparableT]:
        """Find the n largest elements in the dataset."""
        if key is not None:
            return nlargest(n, self.data, key=key)
        else:
            return nlargest(n, self.data)

    def nsmallest(
        self, n: int, key: Optional[Callable] = None
    ) -> list[_ComparableT]:
        """Find the n smallest elements in the dataset."""
        if key is not None:
            return nsmallest(n, self.data, key=key)
        else:
            return nsmallest(n, self.data)

    def insert(self, index: int, value: _ComparableT) -> None:
        self.data.insert(index, value)

    def __str__(self) -> str:
        return str(self.data)

    def __repr__(self) -> str:
        return repr(self.data)

    @overload
    def __getitem__(self, s: int) -> _ComparableT: ...

    @overload
    def __getitem__(self, s: slice) -> MutableSequence[_ComparableT]: ...

    def __getitem__(self, s: Any) -> Any:
        return self.data.__getitem__(s)

    @overload
    def __setitem__(self, s: int, o: _ComparableT) -> None: ...

    @overload
    def __setitem__(self, s: slice, o: Iterable[_ComparableT]) -> None: ...

    def __setitem__(self, s: Any, o: Any) -> None:
        self.data.__setitem__(s, o)

    @overload
    def __delitem__(self, i: int) -> None: ...

    @overload
    def __delitem__(self, i: slice) -> None: ...

    def __delitem__(self, i: Any) -> None:
        self.data.__delitem__(i)

    def __len__(self) -> int:
        return len(self.data)

nlargest(n, key=None)

Find the n largest elements in the dataset.

Source code in mode/utils/collections.py

def nlargest(
    self, n: int, key: Optional[Callable] = None
) -> list[_ComparableT]:
    """Find the n largest elements in the dataset."""
    if key is not None:
        return nlargest(n, self.data, key=key)
    else:
        return nlargest(n, self.data)

nsmallest(n, key=None)

Find the n smallest elements in the dataset.

Source code in mode/utils/collections.py

def nsmallest(
    self, n: int, key: Optional[Callable] = None
) -> list[_ComparableT]:
    """Find the n smallest elements in the dataset."""
    if key is not None:
        return nsmallest(n, self.data, key=key)
    else:
        return nsmallest(n, self.data)

pop(index=0)

Pop the smallest item off the heap.

Maintains the heap invariant.

Source code in mode/utils/collections.py

def pop(self, index: int = 0) -> _ComparableT:
    """Pop the smallest item off the heap.

    Maintains the heap invariant.
    """
    if index == 0:
        return heappop(self.data)
    else:
        raise NotImplementedError(
            "Heap can only pop index 0, please use h.data.pop(index)"
        )

push(item)

Push item onto heap, maintaining the heap invariant.

Source code in mode/utils/collections.py

def push(self, item: _ComparableT) -> None:
    """Push item onto heap, maintaining the heap invariant."""
    heappush(self.data, item)

pushpop(item)

Push item on the heap, then pop and return from the heap.

The combined action runs more efficiently than push followed by a separate call to pop.

Source code in mode/utils/collections.py

def pushpop(self, item: _ComparableT) -> _ComparableT:
    """Push item on the heap, then pop and return from the heap.

    The combined action runs more efficiently than
    `push` followed by a separate call to `pop`.
    """
    return heappushpop(self.data, item)

replace(item)

Pop and return the current smallest value, and add the new item.

This is more efficient than :methpop followed by push, and can be more appropriate when using a fixed-size heap.

Note that the value returned may be larger than item! That constrains reasonable uses of this routine unless written as part of a conditional replacement:

if item > heap[0]:
    item = heap.replace(item)

Source code in mode/utils/collections.py

def replace(self, item: _ComparableT) -> _ComparableT:
    """Pop and return the current smallest value, and add the new item.

    This is more efficient than :meth`pop` followed by `push`,
    and can be more appropriate when using a fixed-size heap.

    Note that the value returned may be larger than item!
    That constrains reasonable uses of this routine unless written as
    part of a conditional replacement:

    ```python
    if item > heap[0]:
        item = heap.replace(item)
    ```
    """
    return heapreplace(self.data, item)

LRUCache

Bases: FastUserDict[KT, VT], MutableMapping[KT, VT], MappingViewProxy

LRU Cache implementation using a doubly linked list to track access.

Parameters:

Name	Type	Description	Default
limit	int	The maximum number of keys to keep in the cache. When a new key is inserted and the limit has been exceeded, the Least Recently Used key will be discarded from the cache.	None
thread_safety	bool	Enable if multiple OS threads are going to access/mutate the cache. Defaults to :const:True on free-threaded builds, where there is no GIL to make unguarded access incidentally safe, and :const:False otherwise (which is what it has always been). It cannot be turned off on a free-threaded build -- see the note below.	None

Note

The backing store is an :class:~collections.OrderedDict rather than a plain :class:dict: evicting the oldest entry is this class's hot path, and popitem(last=False) does it in O(1) via the linked list, where the dict equivalent degrades badly as the cache grows (measured in docs/free-threading.md). The cost of that linked list is that OrderedDict is not safe to mutate concurrently on free-threaded builds -- racing threads can corrupt it badly enough to segfault the interpreter -- so on those builds the mutex is mandatory rather than merely on by default.

Source code in mode/utils/collections.py

class LRUCache(FastUserDict[KT, VT], MutableMapping[KT, VT], MappingViewProxy):
    """LRU Cache implementation using a doubly linked list to track access.

    Arguments:
        limit (int): The maximum number of keys to keep in the cache.
            When a new key is inserted and the limit has been exceeded,
            the *Least Recently Used* key will be discarded from the
            cache.
        thread_safety (bool): Enable if multiple OS threads are going
            to access/mutate the cache.  Defaults to :const:`True` on
            free-threaded builds, where there is no GIL to make unguarded
            access incidentally safe, and :const:`False` otherwise (which
            is what it has always been).  It cannot be turned off on a
            free-threaded build -- see the note below.

    Note:
        The backing store is an :class:`~collections.OrderedDict` rather
        than a plain :class:`dict`: evicting the oldest entry is this
        class's hot path, and ``popitem(last=False)`` does it in O(1) via
        the linked list, where the ``dict`` equivalent degrades badly as
        the cache grows (measured in ``docs/free-threading.md``).  The
        cost of that linked list is that `OrderedDict` is not safe to
        mutate concurrently on free-threaded builds -- racing threads can
        corrupt it badly enough to segfault the interpreter -- so on
        those builds the mutex is mandatory rather than merely on by
        default.
    """

    limit: Optional[int]
    thread_safety: bool
    _mutex: AbstractContextManager
    data: OrderedDict

    def __init__(
        self,
        limit: Optional[int] = None,
        *,
        thread_safety: Optional[bool] = None,
    ) -> None:
        self.limit = limit
        if thread_safety is None:
            thread_safety = FREE_THREADED
        elif FREE_THREADED and not thread_safety:
            # Not a preference we can honour: an unguarded OrderedDict on a
            # free-threaded build is memory-unsafe, not merely racy, and
            # taking the interpreter down is a worse outcome than ignoring
            # the argument.  Say so rather than doing it silently.
            raise ValueError(
                "LRUCache(thread_safety=False) is not supported on "
                "free-threaded builds: the backing OrderedDict can be "
                "corrupted by concurrent mutation badly enough to "
                "segfault the interpreter. Omit the argument to get the "
                "mutex, which is the default here."
            )
        self.thread_safety = thread_safety
        self._mutex = self._new_lock()
        self.data: OrderedDict = OrderedDict()

    def __getitem__(self, key: KT) -> VT:
        with self._mutex:
            return self._touch(key)

    def _touch(self, key: KT) -> VT:
        # Caller must hold the mutex.  Pop and re-insert to mark the key
        # most recently used.
        value = self.data.pop(key)
        self.data[key] = value
        return cast(VT, value)

    def update(self, *args: Any, **kwargs: Any) -> None:
        with self._mutex:
            data, limit = self.data, self.limit
            data.update(*args, **kwargs)
            if limit and len(data) > limit:
                # pop additional items in case limit exceeded
                for _ in range(len(data) - limit):
                    data.popitem(last=False)

    def popitem(self, *, last: bool = True) -> tuple[KT, VT]:
        with self._mutex:
            return self.data.popitem(last)

    def __setitem__(self, key: KT, value: VT) -> None:
        with self._mutex:
            self._store(key, value)

    def _store(self, key: KT, value: VT) -> None:
        # Caller must hold the mutex.  Evict the least recently used
        # entry when inserting a *new* key into a full cache -- updating
        # an existing key does not grow the cache, so evicting for it
        # would shrink a full cache on every such update.  The cheap size
        # checks run first so unlimited and under-limit caches skip the
        # containment probe.  Eviction is a single `popitem(last=False)`
        # call, NOT `pop(next(iter(...)))`, so that even an unlocked
        # cache on a GIL build (the historical default there) cannot race
        # the iter/next/pop gaps; docs/free-threading.md has the story.
        if (
            self.limit
            and len(self.data) >= self.limit
            and key not in self.data
        ):
            self.data.popitem(last=False)
        self.data[key] = value

    # NOTE: Iteration operates on a `copy()` snapshot taken under the
    # mutex, iterated with the mutex released.  Holding the lock across
    # yields would keep it held for as long as the *consumer* takes --
    # forever, if a generator is abandoned half-consumed -- and iterating
    # the live dict is what "changed size during iteration" used to be.

    def __iter__(self) -> Iterator:
        return self._keys()

    def keys(self) -> KeysView[KT]:
        return ProxyKeysView(self)

    def _keys(self) -> Iterator[KT]:
        return iter(self.copy())

    def values(self) -> ValuesView[VT]:
        return ProxyValuesView(self)

    def _values(self) -> Iterator[VT]:
        return iter(self.copy().values())

    def items(self) -> ItemsView[KT, VT]:
        return ProxyItemsView(self)

    def _items(self) -> Iterator[tuple[KT, VT]]:
        return iter(self.copy().items())

    def incr(self, key: KT, delta: int = 1) -> int:
        with self._mutex:
            # this acts as memcached does- store as a string, but return a
            # integer as long as it exists and we can cast it
            newval = int(self.data.pop(key)) + delta
            self._store(key, cast(VT, str(newval)))
            return newval

    # NOTE: Everything below re-implements an inherited method that would
    # otherwise reach `self.data` with the mutex released.  On a
    # free-threaded build that is not merely a stale answer: touching the
    # backing OrderedDict while another thread mutates it can corrupt its
    # linked list badly enough to take the interpreter down, and a
    # read-only operation such as `len` or `in` is just as capable of
    # observing the half-updated state as a write is.  Anything added to
    # `FastUserDict` that uses `self.data` directly needs an override here
    # too.

    def __delitem__(self, key: KT) -> None:
        with self._mutex:
            del self.data[key]

    def __len__(self) -> int:
        with self._mutex:
            return len(self.data)

    def __contains__(self, key: object) -> bool:
        with self._mutex:
            return key in self.data

    def __repr__(self) -> str:
        with self._mutex:
            return repr(self.data)

    def copy(self) -> dict:
        with self._mutex:
            return dict(self.data)

    def clear(self) -> None:
        with self._mutex:
            self.data.clear()

    # `pop` and `setdefault` below are inherited as combinations of the
    # locked primitives above, which is already memory-safe -- but the
    # lock is dropped between the lookup and the store, a surprising
    # place for a class that advertises thread safety to lose an
    # invariant.  They are made atomic instead.  (`get` needs no
    # override: its only data access is the single, already-locked
    # `self[key]`.)

    @overload
    def pop(self, key: KT) -> VT: ...

    @overload
    def pop(self, key: KT, default: Union[VT, T]) -> Union[VT, T]: ...

    def pop(self, key: KT, default: Any = _MISSING) -> Any:
        with self._mutex:
            try:
                return cast(VT, self.data.pop(key))
            except KeyError:
                if default is _MISSING:
                    raise
                return default

    def setdefault(self, key: KT, default: Any = None) -> Any:
        with self._mutex:
            try:
                return self._touch(key)
            except KeyError:
                self._store(key, cast(VT, default))
                return default

    def _new_lock(self) -> AbstractContextManager:
        if self.thread_safety:
            return cast(AbstractContextManager, threading.RLock())
        return cast(AbstractContextManager, nullcontext())

    def __getstate__(self) -> Mapping[str, Any]:
        d = dict(vars(self))
        d.pop("_mutex")
        return d

    def __setstate__(self, state: dict[str, Any]) -> None:
        # Unpickling is another way to construct the object, so it has to
        # honour the same invariant `__init__` does: no unguarded
        # OrderedDict on a free-threaded build.  Pickles written by an
        # older version -- or on a GIL build, where thread_safety=False is
        # both the default and legal -- would otherwise come back here
        # with `nullcontext` for a mutex.  Upgrading the flag keeps those
        # pickles loadable, which raising would not.
        if FREE_THREADED and not state.get("thread_safety", False):
            state["thread_safety"] = True
        self.__dict__ = state
        self._mutex = self._new_lock()

ManagedUserDict

Bases: FastUserDict[KT, VT]

A UserDict that adds callbacks for when keys are get/set/deleted.

Source code in mode/utils/collections.py

class ManagedUserDict(FastUserDict[KT, VT]):
    """A UserDict that adds callbacks for when keys are get/set/deleted."""

    def on_key_get(self, key: KT) -> None:
        """Handle that key is being retrieved."""
        ...

    def on_key_set(self, key: KT, value: VT) -> None:
        """Handle that value for a key is being set."""
        ...

    def on_key_del(self, key: KT) -> None:
        """Handle that a key is deleted."""
        ...

    def on_clear(self) -> None:
        """Handle that the mapping is being cleared."""
        ...

    def __getitem__(self, key: KT) -> Any:
        self.on_key_get(key)
        return super().__getitem__(key)

    def __setitem__(self, key: KT, value: VT) -> None:
        self.on_key_set(key, value)
        self.data[key] = value

    def __delitem__(self, key: KT) -> None:
        self.on_key_del(key)
        del self.data[key]

    def update(self, *args: Any, **kwargs: Any) -> None:
        for d in args:
            for key, value in d.items():
                self.on_key_set(key, value)
        for key, value in kwargs.items():
            self.on_key_set(cast(KT, key), value)
        self.data.update(*args, **kwargs)

    def clear(self) -> None:
        self.on_clear()
        self.data.clear()

    def raw_update(self, *args: Any, **kwargs: Any) -> None:
        self.data.update(*args, **kwargs)

on_clear()

Handle that the mapping is being cleared.

Source code in mode/utils/collections.py

def on_clear(self) -> None:
    """Handle that the mapping is being cleared."""
    ...

on_key_del(key)

Handle that a key is deleted.

Source code in mode/utils/collections.py

def on_key_del(self, key: KT) -> None:
    """Handle that a key is deleted."""
    ...

on_key_get(key)

Handle that key is being retrieved.

Source code in mode/utils/collections.py

def on_key_get(self, key: KT) -> None:
    """Handle that key is being retrieved."""
    ...

on_key_set(key, value)

Handle that value for a key is being set.

Source code in mode/utils/collections.py

def on_key_set(self, key: KT, value: VT) -> None:
    """Handle that value for a key is being set."""
    ...

ManagedUserSet

Bases: FastUserSet[T]

A MutableSet that adds callbacks for when keys are get/set/deleted.

Source code in mode/utils/collections.py

class ManagedUserSet(FastUserSet[T]):
    """A MutableSet that adds callbacks for when keys are get/set/deleted."""

    def on_add(self, value: T) -> None: ...

    def on_discard(self, value: T) -> None: ...

    def on_clear(self) -> None: ...

    def on_change(self, added: set[T], removed: set[T]) -> None: ...

    def add(self, element: T) -> None:
        if element not in self.data:
            self.on_add(element)
            self.data.add(element)

    def clear(self) -> None:
        self.on_clear()
        self.data.clear()

    def discard(self, element: T) -> None:
        if element in self.data:
            self.on_discard(element)
            self.data.discard(element)

    def pop(self) -> T:
        element = self.data.pop()
        self.on_discard(element)
        return element

    def raw_update(self, *args: Any, **kwargs: Any) -> None:
        self.data.update(*args, **kwargs)  # type: ignore

    def __iand__(self, other: Set[Any]) -> "ManagedUserSet[T]":
        self.on_change(added=set(), removed=cast(set, self).difference(other))
        self.data.__iand__(other)
        return self

    def __ior__(self, other: Set[_S]) -> "ManagedUserSet[T]":
        self.on_change(added=cast(set, other).difference(self), removed=set())
        cast(MutableSet[Union[T, _S]], self.data).__ior__(other)
        return self

    def __isub__(self, other: Set[Any]) -> "ManagedUserSet[T]":
        self.on_change(
            added=set(), removed=cast(set, self.data).intersection(other)
        )
        self.data.__isub__(other)
        return self

    def __ixor__(self, other: Set[_S]) -> "ManagedUserSet[T]":
        self.on_change(
            added=cast(set, other).difference(self.data),
            removed=cast(set, self.data).intersection(other),
        )
        cast(MutableSet[Union[T, _S]], self.data).__ixor__(other)
        return self

    def difference_update(self, other: _Setlike[T]) -> None:
        data = cast(set, self.data)
        self.on_change(added=set(), removed=data.intersection(other))
        data.difference_update(other)

    def intersection_update(self, other: _Setlike[T]) -> None:
        data = cast(set, self.data)
        self.on_change(added=set(), removed=cast(set, self).difference(other))
        data.intersection_update(other)

    def symmetric_difference_update(self, other: _Setlike[T]) -> None:
        data = cast(set, self.data)
        self.on_change(
            added=cast(set, other).difference(self.data),
            removed=data.intersection(other),
        )
        data.symmetric_difference_update(other)

    def update(self, other: _Setlike[T]) -> None:
        # union update
        self.on_change(added=cast(set, other).difference(self), removed=set())
        cast(set, self.data).update(other)

force_mapping(m)

Wrap object into supporting the mapping interface if necessary.

Source code in mode/utils/collections.py

def force_mapping(m: Any) -> Mapping:
    """Wrap object into supporting the mapping interface if necessary."""
    if isinstance(m, (LazyObject, LazySettings)):
        m = m._wrapped
    return DictAttribute(m) if not isinstance(m, Mapping) else m