| 54 | |
| 55 | |
| 56 | class ObservableList(UserList): |
| 57 | def __init__(self, initlist: Optional[Iterable[_T]] = None) -> None: |
| 58 | super().__init__() |
| 59 | self._observers = [] |
| 60 | |
| 61 | if initlist: |
| 62 | self.extend(initlist) |
| 63 | |
| 64 | def subscribe(self, observer): |
| 65 | self._observers.append(observer) |
| 66 | |
| 67 | def unsubscribe(self, observer): |
| 68 | self._observers.remove(observer) |
| 69 | |
| 70 | def append(self, item: _T) -> None: |
| 71 | super().append(item) |
| 72 | |
| 73 | for observer in self._observers: |
| 74 | observer.on_add(item, len(self.data) - 1) |
| 75 | |
| 76 | def insert(self, i: int, item: _T) -> None: |
| 77 | super().insert(i, item) |
| 78 | |
| 79 | for observer in self._observers: |
| 80 | observer.on_add(item, i) |
| 81 | |
| 82 | def pop(self, i: int = ...) -> _T: |
| 83 | item = super().pop(i) |
| 84 | |
| 85 | for observer in self._observers: |
| 86 | observer.on_remove(item) |
| 87 | |
| 88 | return item |
| 89 | |
| 90 | def remove(self, item: _T) -> None: |
| 91 | super().remove(item) |
| 92 | |
| 93 | for observer in self._observers: |
| 94 | observer.on_remove(item) |
| 95 | |
| 96 | def clear(self) -> None: |
| 97 | copy = list(self.data) |
| 98 | |
| 99 | super().clear() |
| 100 | |
| 101 | for item in copy: |
| 102 | for observer in self._observers: |
| 103 | observer.on_remove(item) |
| 104 | |
| 105 | def extend(self, other: Iterable[_T]) -> None: |
| 106 | first_index = len(self.data) |
| 107 | |
| 108 | super().extend(other) |
| 109 | |
| 110 | for i, item in enumerate(other): |
| 111 | for observer in self._observers: |
| 112 | observer.on_add(item, first_index + i) |
| 113 |
no outgoing calls