Class ObservableDict

src/react/properties.py:115–178

class ObservableDict(UserDict):
    def __init__(self, dict: Optional[Mapping[_KT, _VT]] = None, **kwargs: _VT) -> None:
        super().__init__(**kwargs)
        self._observers = []

        if dict:
            self.update(dict)

    def subscribe(self, observer):
        self._observers.append(observer)

    def unsubscribe(self, observer):
        if observer in self._observers:
            self._observers.remove(observer)

    def set(self, another_dict):
        old_values = dict(self)

        obsolete_keys = {key for key in self.keys() if key not in another_dict}
        for key in obsolete_keys:
            super().__delitem__(key)

        for key, value in another_dict.items():
            super().__setitem__(key, value)

        if self._observers:
            for obsolete_key in obsolete_keys:
                old_value = old_values[obsolete_key]
                for observer in self._observers:
                    observer(obsolete_key, old_value, None)

            for key, value in self.items():
                old_value = old_values.get(key)
                if old_value != value:
                    for observer in self._observers:
                        observer(key, old_value, value)

    def __setitem__(self, key: _KT, item: _VT) -> None:
        old_value = self.get(key)

        super().__setitem__(key, item)

        if item == old_value:
            return

        self._notify_observers(key, old_value, item)

    def __delitem__(self, key: _KT) -> None:
        old_value = self.get(key)

        super().__delitem__(key)

        if old_value is None:
            return

        self._notify_observers(key, old_value, None)

    def _notify_observers(self, key, old_value, new_value):