Исходный код EJIO.ctk_extender.state.collection

## /ctk_extender/state/collection.py

from collections.abc import Iterable, Iterator
from typing import overload, Any
from EJIO.utils import List, Dict
from ..event import EventController
from .events import CollectionChangedEvent, DictionaryChangedEvent

__all__ = [
    'ReactiveList',
    'ReactiveDict',
]




class ReactiveList[T]:
    """
    Реактивная обертка над потокобезопасным двусвязным списком.
    Любое изменение состава элементов порождает CollectionChangedEvent.
    """
    __slots__ = ('_storage',)

    def __init__(self, iterable: Iterable[T] | None = None) -> None:
        self._storage: List[T] = List(iterable)



    def append(self, item: T) -> None:
        """Добавить элемент в конец коллекции."""
        self._storage.append(item)
        EventController.emit(CollectionChangedEvent(self, 'append', [item]))




    def extend(self, iterable: Iterable[T]) -> None:
        """Добавить все элементы из итерируемого объекта."""
        items_list = list(iterable)
        self._storage.extend(items_list)
        EventController.emit(CollectionChangedEvent(self, 'extend', items_list))




    def remove(self, item: T) -> None:
        """Удалить первое вхождение элемента."""
        self._storage.remove(item)
        EventController.emit(CollectionChangedEvent(self, 'remove', [item]))




    def pop(self, index: int = -1) -> T:
        """Удалить и вернуть элемент по индексу."""
        removed_item = self._storage.pop(index)
        EventController.emit(CollectionChangedEvent(self, 'pop', [removed_item]))
        return removed_item




    def clear(self) -> None:
        """Очистить коллекцию."""
        # Перед очисткой создаем снимок элементов для события
        snapshot = list(self._storage)
        self._storage.clear()
        EventController.emit(CollectionChangedEvent(self, 'clear', snapshot))


    @overload
    def __getitem__(self, index: int) -> T: ...

    @overload
    def __getitem__(self, index: slice) -> List[T]: ...

    def __getitem__(self, index: int | slice) -> T | List[T]:
        return self._storage[index]

    def __setitem__(self, index: int, value: T) -> None:
        self._storage[index] = value
        EventController.emit(CollectionChangedEvent(self, 'set', [value]))

    def __delitem__(self, index: int) -> None:
        removed_item = self._storage[index]
        del self._storage[index]
        EventController.emit(CollectionChangedEvent(self, 'pop', [removed_item]))

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

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

    def __repr__(self) -> str:
        return f"ReactiveList({str(self._storage)})"

    def __del__(self) -> None:
        """
        Деструктор коллекции. Принудительно очищает внутреннее потокобезопасное
        хранилище для предотвращения застревания объектов в памяти.
        """
        try:
            self._storage.clear()
        except Exception:
            pass





class ReactiveDict[K, V]:
    """Реактивная событийная обертка над потокобезопасным словарем."""
    __slots__ = ('_storage',)

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        self._storage: Dict[K, V] = Dict(*args, **kwargs)

    def __getitem__(self, key: K) -> V:
        return self._storage[key]

    def __setitem__(self, key: K, value: V) -> None:
        self._storage[key] = value
        EventController.emit(DictionaryChangedEvent(self, 'set', key, value))

    def __delitem__(self, key: K) -> None:
        old_value = self._storage[key]
        del self._storage[key]
        EventController.emit(DictionaryChangedEvent(self, 'delete', key, old_value))



    def clear(self) -> None:
        self._storage.clear()
        EventController.emit(DictionaryChangedEvent(self, 'clear', None, None))


    def __iter__(self) -> Iterator[K]:
        return iter(self._storage)

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

    def __repr__(self) -> str:
        return f"ReactiveDict({self._storage!r})"