# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Module for managing a store of identifiable items with event notifications.
This module provides the :class:`ItemStore` class that allows storing
identifiable items and notifying listeners about changes to the store.
It also defines the :class:`StoreEvent` enum to represent different types
of events that can occur in the store.
"""
from __future__ import annotations
import sys
if sys.version_info >= (3, 11):
from enum import StrEnum
else:
from radioviz.models.legacy import StrEnum
from collections.abc import Iterator
from typing import Any, Callable, Generic, Protocol, Sequence, TypeVar, Union, overload
from uuid import UUID
[docs]
class StoreEvent(StrEnum):
"""
Enumeration of possible store events.
This enum defines the different types of events that can be triggered
when items are added, removed, updated, or the store is cleared.
"""
ADDED = 'added'
"""Item was added to the store."""
REMOVED = 'removed'
"""Item was removed from the store."""
UPDATED = 'updated'
"""Item was updated in the store."""
CLEARED = 'cleared'
"""Store was cleared of all items."""
[docs]
class IdentifiableItem(Protocol):
"""
Protocol for identifiable items.
Any item that implements this protocol must have an `id` attribute
of type :class:`UUID`.
"""
id: UUID
"""Unique identifier of the item."""
T = TypeVar('T', bound=IdentifiableItem)
[docs]
class ItemStore(Sequence[T], Generic[T]):
"""
A generic store for identifiable items with event notifications.
This class manages a collection of items that implement the
:class:`IdentifiableItem` protocol. It supports adding, removing,
and retrieving items, as well as notifying registered listeners
about changes to the store through :class:`StoreEvent` notifications.
"""
def __init__(self) -> None:
"""
Initialize an empty item store.
Creates a new instance with an empty list of items and no listeners.
"""
self._items: list[T] = []
self._listeners: list[Callable[[StoreEvent, T | None, int], None]] = []
[docs]
def add_listener(self, callback: Callable[[StoreEvent, T | None, int], None]) -> None:
"""
Add a listener to receive store change notifications.
:param callback: A callable that will be invoked when store events occur.
The callable should accept three parameters:
(event: StoreEvent, item: T | None, index: int)
:type callback: Callable[[StoreEvent, T | None, int], None]
"""
self._listeners.append(callback)
[docs]
def add_update_listener(self, callback: Callable[[T, int], None]) -> None:
"""
Add a listener for update events with a guaranteed item.
:param callback: A callable invoked on update events with (item, index).
:type callback: Callable[[T, int], None]
"""
def _handler(event: StoreEvent, item: T | None, index: int) -> None:
if event == StoreEvent.UPDATED and item is not None:
callback(item, index)
self._listeners.append(_handler)
[docs]
def _notify(self, event: StoreEvent, item_data: T | None, index: int) -> None:
"""
Notify all registered listeners about a store event.
:param event: The type of event that occurred.
:type event: StoreEvent
:param item_data: The item data associated with the event, if applicable.
:type item_data: T | None
:param index: The index of the item in the store, if applicable.
:type index: int
"""
for cb in self._listeners:
cb(event, item_data, index)
[docs]
def add(self, item: T) -> None:
"""
Add an item to the store.
:param item: The item to add to the store.
:type item: T
"""
self._items.append(item)
self._notify(StoreEvent.ADDED, item, len(self._items) - 1)
def update_by_id(self, item_id: UUID, **changes: Any) -> None:
item = self.get_by_id(item_id)
for key, value in changes.items():
setattr(item, key, value)
self._notify(StoreEvent.UPDATED, item, self.index_of(item_id))
[docs]
def remove(self, index: int) -> None:
"""
Remove an item from the store by index.
:param index: The index of the item to remove.
:type index: int
:raise IndexError: if the index is out of range.
"""
profile = self._items.pop(index)
self._notify(StoreEvent.REMOVED, profile, index)
[docs]
def remove_item(self, item: T) -> None:
"""
Remove an item from the store by reference.
:param item: The item to remove from the store.
:type item: T
"""
for i, p in enumerate(self._items):
if p == item:
self.remove(i)
break
[docs]
def reset(self) -> None:
"""
Clear all items from the store.
This method removes all items from the store and notifies listeners
of the :attr:`StoreEvent.CLEARED` event.
"""
self._items.clear()
self._notify(StoreEvent.CLEARED, None, -1)
[docs]
def index_of(self, item_id: UUID) -> int:
"""
Find the index of an item by its ID.
:param item_id: The unique identifier of the item to find.
:type item_id: UUID
:return: The index of the item in the store.
:rtype: int
:raise KeyError: if no item with the specified ID exists.
"""
for i, item in enumerate(self._items):
if item.id == item_id:
return i
raise KeyError(f'No item with id {item_id}')
[docs]
def get_by_id(self, item_id: UUID) -> T:
"""
Retrieve an item from the store by its ID.
:param item_id: The unique identifier of the item to retrieve.
:type item_id: UUID
:return: The item with the specified ID.
:rtype: T
"""
return self._items[self.index_of(item_id)]
@overload
def __getitem__(self, index: int) -> T: ...
@overload
def __getitem__(self, index: slice) -> Sequence[T]: ...
def __getitem__(self, i: Union[int, slice]) -> Union[T, Sequence[T]]:
"""
Get an item from the store by index.
:param i: The index of the item to retrieve.
:type i: int or slice
:return: The item at the specified index.
:rtype: T or Sequence[T]
"""
if isinstance(i, slice):
return self._items[i]
return self._items[i]
[docs]
def get(self, index: int) -> Union[T, None]:
"""
Safely retrieve an item from the store by index.
:param index: The index of the item to retrieve.
:type index: int
:return: The item at the specified index, or None if index is out of range.
:rtype: T | None
"""
if 0 <= index < len(self._items):
return self._items[index]
return None
def __iter__(self) -> Iterator[T]:
return iter(self._items)
[docs]
def all(self) -> list[T]:
"""
Get all items in the store.
:return: A copy of the list of all items in the store.
:rtype: list[T]
"""
return list(self._items)
def __len__(self) -> int:
"""
Get the number of items in the store.
:return: The number of items in the store.
:rtype: int
"""
return len(self._items)