from typing import List, Iterable, Dict, TypeVar, Generic, Iterator from collections import defaultdict from dataclasses import dataclass from .parents import DatabaseObject from ..utils.hooks import HookEventTypes, Hooks, Event class CollectionHooks(HookEventTypes): APPEND_NEW = "append_new" T = TypeVar('T', bound=DatabaseObject) @dataclass class AppendResult: was_in_collection: bool current_element: DatabaseObject was_the_same: bool class Collection(Generic[T]): """ This a class for the iterables like tracklist or discography """ _data: List[T] _by_url: dict _by_attribute: dict def __init__(self, data: List[T] = None, element_type=None, *args, **kwargs) -> None: # Attribute needs to point to self.element_type = element_type self._data: List[T] = list() """ example of attribute_to_object_map the song objects are references pointing to objects in _data ```python { 'id': {323: song_1, 563: song_2, 666: song_3}, 'url': {'www.song_2.com': song_2} } ``` """ self._attribute_to_object_map: Dict[str, Dict[object, T]] = defaultdict(dict) self._used_ids: set = set() self.hooks: Hooks = Hooks(self) if data is not None: self.extend(data, merge_on_conflict=True) def sort(self, reverse: bool = False, **kwargs): self._data.sort(reverse=reverse, **kwargs) def map_element(self, element: T): for name, value in element.indexing_values: if value is None: continue self._attribute_to_object_map[name][value] = element self._used_ids.add(element.id) def unmap_element(self, element: T): for name, value in element.indexing_values: if value is None: continue if value in self._attribute_to_object_map[name]: if element is self._attribute_to_object_map[name][value]: try: self._attribute_to_object_map[name].pop(value) except KeyError: pass def append(self, element: T, merge_on_conflict: bool = True, merge_into_existing: bool = True, no_hook: bool = False) -> AppendResult: """ :param element: :param merge_on_conflict: :param merge_into_existing: :return did_not_exist: """ if element is None: return AppendResult(False, None, False) for existing_element in self._data: if element is existing_element: return AppendResult(False, None, False) # if the element type has been defined in the initializer it checks if the type matches if self.element_type is not None and not isinstance(element, self.element_type): raise TypeError(f"{type(element)} is not the set type {self.element_type}") # return if the same instance of the object is in the list for existing in self._data: if element is existing: return AppendResult(True, element, True) for name, value in element.indexing_values: if value in self._attribute_to_object_map[name]: existing_object = self._attribute_to_object_map[name][value] if not merge_on_conflict: return AppendResult(True, existing_object, False) # if the object does already exist # thus merging and don't add it afterward if merge_into_existing: existing_object.merge(element) # in case any relevant data has been added (e.g. it remaps the old object) self.map_element(existing_object) return AppendResult(True, existing_object, False) element.merge(existing_object) exists_at = self._data.index(existing_object) self._data[exists_at] = element self.unmap_element(existing_object) self.map_element(element) return AppendResult(True, existing_object, False) if not no_hook: self.hooks.trigger_event(CollectionHooks.APPEND_NEW, new_object=element) self._data.append(element) self.map_element(element) return AppendResult(False, element, False) def extend(self, element_list: Iterable[T], merge_on_conflict: bool = True, merge_into_existing: bool = True, no_hook: bool = False): if element_list is None: return if len(element_list) <= 0: return if element_list is self: return for element in element_list: self.append(element, merge_on_conflict=merge_on_conflict, merge_into_existing=merge_into_existing, no_hook=no_hook) def sync_collection(self, collection_attribute: str): def on_append(event: Event, new_object: T, *args, **kwargs): new_collection = new_object.__getattribute__(collection_attribute) if self is new_collection: return self.extend(new_object.__getattribute__(collection_attribute), no_hook=True) new_object.__setattr__(collection_attribute, self) self.hooks.add_event_listener(CollectionHooks.APPEND_NEW, on_append) def sync_main_collection(self, main_collection: "Collection", collection_attribute: str): def on_append(event: Event, new_object: T, *args, **kwargs): new_collection = new_object.__getattribute__(collection_attribute) if main_collection is new_collection: return main_collection.extend(new_object.__getattribute__(collection_attribute), no_hook=True) new_object.__setattr__(collection_attribute, main_collection) self.hooks.add_event_listener(CollectionHooks.APPEND_NEW, on_append) """ def on_append(event: Event, new_object: T, *args, **kwargs): new_collection: Collection = new_object.__getattribute__(collection_attribute) if self is new_collection: return self.extend(new_collection.shallow_list, no_hook=False) new_object.__setattr__(collection_attribute, self) self.hooks.add_event_listener(CollectionHooks.APPEND_NEW, on_append) """ def __iter__(self) -> Iterator[T]: for element in self._data: yield element def __str__(self) -> str: return "\n".join([f"{str(j).zfill(2)}: {i.__repr__()}" for j, i in enumerate(self._data)]) def __len__(self) -> int: return len(self._data) def __getitem__(self, key) -> T: if type(key) != int: return ValueError("key needs to be an integer") return self._data[key] def __setitem__(self, key, value: T): if type(key) != int: return ValueError("key needs to be an integer") old_item = self._data[key] self.unmap_element(old_item) self.map_element(value) self._data[key] = value @property def shallow_list(self) -> List[T]: """ returns a shallow copy of the data list """ return self._data.copy() @property def empty(self) -> bool: return len(self._data) == 0 def clear(self): self.__init__(element_type=self.element_type)