Concurrent List
- class thread_factory.concurrency.concurrent_list.ConcurrentList(initial: Iterable[_T] | None = None)[source]
Bases:
Generic[_T],IDisposableA thread-safe list implementation using an underlying Python list, a reentrant lock for synchronization, and an atomic counter for fast, lock-free retrieval of the length.
This class mimics many of the behaviors of a native Python list, including slicing, in-place operators, and common utility methods. It is designed for Python 3.13+ No-GIL environments.
- append(item: _T) None[source]
Append an item to the end of the list.
- Parameters:
item (_T) – The item to append.
- batch_update(func: Callable[[List[_T]], None]) None[source]
Perform a batch update on the list under a single lock acquisition. This method allows multiple operations to be performed atomically.
- Parameters:
func (Callable[[List[_T]], None]) – A function that accepts the internal list as its only argument. The function should perform all necessary mutations.
- copy() ConcurrentList[_T][source]
Return a shallow copy of the ConcurrentList.
- Returns:
A new ConcurrentList with copied items.
- Return type:
ConcurrentList[_T]
- count(item: Any) int[source]
Return the number of occurrences of a value.
- Parameters:
item (Any) – The item to count.
- Returns:
The number of occurrences.
- Return type:
int
- dispose() None[source]
Dispose (clear) this ConcurrentList, releasing its contents.
Once disposed, _disposed becomes True and the internal dict is cleared. No further usage checks are enforced, so the user must avoid calling other methods after disposal.
This method is idempotent — multiple calls won’t cause errors.
- extend(items: Iterable[_T]) None[source]
Extend the list by appending elements from the iterable.
- Parameters:
items (Iterable[_T]) – An iterable of items to add.
- Raises:
TypeError – If items is not iterable (e.g., if it’s None).
- filter(func: Callable[[_T], bool]) ConcurrentList[_T][source]
Filter elements based on a function and return a new ConcurrentList.
- Parameters:
func (Callable[[_T], bool]) – The filter function.
- Returns:
A new ConcurrentList containing only elements where func(element) is True.
- Return type:
ConcurrentList[_T]
- freeze() None[source]
Freeze the dictionary to prevent further modifications. This is useful for making the dictionary immutable after initialization.
- index(item: Any, start: int = 0, end: int | None = None) int[source]
Return first index of value.
- Parameters:
item (Any) – The item to find.
start (int, optional) – Start index for search.
end (int, optional) – End index for search.
- Returns:
The index of the item.
- Return type:
int
- Raises:
ValueError – If the item is not present.
- insert(index: int, item: _T) None[source]
Insert an item at the specified index.
- Parameters:
index (int) – The index at which to insert.
item (_T) – The item to insert.
- Raises:
IndexError – If the index is out of range (depending on desired behavior).
- property is_frozen: bool
Check if the dictionary is frozen.
- Returns:
True if the dictionary is frozen, False otherwise.
- Return type:
bool
- map(func: Callable[[_T], Any]) ConcurrentList[Any][source]
Apply a function to all elements and return a new ConcurrentList.
- Parameters:
func (Callable[[_T], Any]) – The function to apply.
- Returns:
A new ConcurrentList with the function applied to each element.
- Return type:
ConcurrentList[Any]
- pop(index: int = -1) _T[source]
Remove and return the item at the given index (default is last).
- Parameters:
index (int, optional) – The index to pop. Defaults to -1.
- Returns:
The popped item.
- Return type:
_T
- Raises:
IndexError – If the list is empty or index is out of range.
- reduce(func: Callable[[Any, _T], Any], initial: Any | None = None) Any[source]
Apply a function of two arguments cumulatively to the items of the list.
- Parameters:
func (Callable[[Any, _T], Any]) – Function to apply.
initial (Any, optional) – Starting value.
- Returns:
The reduced value.
- Return type:
Any
- Raises:
TypeError – If the list is empty and no initial value is provided.
- remove(item: _T) None[source]
Remove the first occurrence of an item from the list.
- Parameters:
item (_T) – The item to remove.
- Raises:
ValueError – If the item is not found.
- sort(key: Callable[[_T], Any] | None = None, reverse: bool = False) None[source]
Sort the list in-place.
- Parameters:
key (Callable[[_T], Any], optional) – A function used to extract a comparison key.
reverse (bool, optional) – If True, the list elements are sorted as if each comparison were reversed.
- to_list() List[_T][source]
Return a shallow copy of the internal list.
- Returns:
A copy of the list.
- Return type:
List[_T]