## /utils/struct/list.py
from collections.abc import Callable, Iterable
from threading import Lock
from typing import Self, overload
__all__ = [
'List',
]
class Node[T]:
"""
Узел двусвязного списка для хранения значения и указателей на соседей.
"""
__slots__ = ('value', 'prev', 'next')
def __init__(
self,
value: T | None = None,
prev: 'Node[T] | None' = None,
next: 'Node[T] | None' = None
) -> None:
"""
Создает узел двусвязного списка.
:param value: Значение записываемое в узел.
:param next: Ссылка на следующий узел.
:param prev: Ссылка на предыдущий узел.
"""
self.value: T | None = value
self.prev: 'Node[T] | None' = prev
self.next: 'Node[T] | None' = next
class ListIterator[T]:
"""
Итератор по списку, устойчивый к удалению элементов во время обхода.
"""
__slots__ = ('_current', '_end', '_lock')
def __init__(
self,
start_node: Node[T],
end_node: Node[T],
lock: Lock
) -> None:
"""
Создает итератор по потокобезопасному списку.
:param start_node: Начальный узел итератора.
:param end_node: Конечный узел итератора.
:param lock: Замок потокобезопасного списка.
"""
self._current: Node[T] = start_node
self._end: Node[T] = end_node
self._lock: Lock = lock
def __iter__(self) -> Self:
"""
Возвращает итератор потокобезопасного списка.
:return: Итератор потокобезопасного списка.
"""
return self
def __next__(self) -> T:
"""
Возвращает следующее значение в потокобезопасном списке.
:return: Значение узла.
:raises: StopIteration в конце обхода.
"""
with self._lock:
if self._current is self._end or self._current.next is self._end:
raise StopIteration
self._current = self._current.next # type: ignore
return self._current.value # type: ignore
[документация]
class List[T]:
"""
Потокобезопасный динамический список на базе двусвязного списка.
Итерация по списку безопасна к удалениям элементов "на лету".
"""
__slots__ = ('_head', '_tail', '_size', '_lock')
def __init__(self, iterable: Iterable[T] | None = None) -> None:
"""
Создает объект потокобехопасного списка.
:param iterable: Перечисляемый объект из которо следует создать список.
"""
self._lock: Lock = Lock()
self._size: int = 0
# Фиктивные (dummy) узлы для удобства вставки/удаления на краях
self._head: Node[T] = Node()
self._tail: Node[T] = Node()
self._head.next = self._tail
self._tail.prev = self._head
if iterable is not None:
self.extend(iterable)
def _get_node(self, index: int) -> Node[T]:
"""
Внутренний метод получения узла по индексу (поддерживает отрицательные индексы).
Вызывать строго внутри блока `with self._lock`.
:param index: Индекс искомого узла.
:return: Узел Node[T].
:raises IndexError: Если индекс находится вне диапазона списка.
"""
if index < 0:
index += self._size
if index < 0 or index >= self._size:
raise IndexError("List index out of range")
# Оптимизация обхода: ищем с начала или с конца в зависимости от индекса
if index < self._size // 2:
current = self._head.next
for _ in range(index):
current = current.next # type: ignore
else:
current = self._tail.prev
for _ in range(self._size - 1 - index):
current = current.prev # type: ignore
return current # type: ignore
[документация]
def append(self, item: T) -> None:
"""
Добавляет элемент в конец списка.
:param item: Элемент для добавления.
"""
with self._lock:
last = self._tail.prev
new_node = Node(item, prev=last, next=self._tail)
last.next = new_node # type: ignore
self._tail.prev = new_node
self._size += 1
[документация]
def extend(self, iterable: Iterable[T]) -> None:
"""
Добавляет все элементы из итерируемого объекта в конец списка.
:param iterable: Итерируемый объект с элементами.
"""
with self._lock:
for item in iterable:
last = self._tail.prev
new_node = Node(item, prev=last, next=self._tail)
last.next = new_node # type: ignore
self._tail.prev = new_node
self._size += 1
[документация]
def insert(self, index: int, item: T) -> None:
"""
Вставляет элемент по указанному индексу.
:param index: Индекс, на место которого будет вставлен элемент.
:param item: Значение для вставки.
"""
with self._lock:
if index < 0:
index += self._size
if index < 0:
index = 0
if index >= self._size:
next_node = self._tail
else:
next_node = self._get_node(index)
prev_node = next_node.prev
new_node = Node(item, prev=prev_node, next=next_node)
prev_node.next = new_node # type: ignore
next_node.prev = new_node
self._size += 1
[документация]
def remove(self, item: T) -> None:
"""
Удаляет первое вхождение элемента. Поддерживает стабильность итератора.
:param item: Элемент, который требуется удалить.
:raises ValueError: Если элемент отсутствует в списке.
"""
with self._lock:
current = self._head.next
while current is not self._tail:
if current.value == item: # type: ignore
current.prev.next = current.next # type: ignore
current.next.prev = current.prev # type: ignore
self._size -= 1
return
current = current.next # type: ignore
raise ValueError(f"List.remove(x): x not in list")
[документация]
def pop(self, index: int = -1) -> T:
"""
Удаляет и возвращает элемент по индексу.
:param index: Индекс удаляемого элемента (по умолчанию последний: -1).
:return: Удаленный элемент списка.
:raises IndexError: Если список пуст или индекс вне диапазона.
"""
with self._lock:
if self._size == 0:
raise IndexError("pop from empty list")
node = self._get_node(index)
node.prev.next = node.next # type: ignore
node.next.prev = node.prev # type: ignore
self._size -= 1
return node.value # type: ignore
[документация]
def clear(self) -> None:
"""Очищает список, удаляя все элементы."""
with self._lock:
current = self._head.next
while current is not self._tail:
next_node = current.next # type: ignore
current.prev = None # type: ignore
current.next = None # type: ignore
current = next_node
self._head.next = self._tail
self._tail.prev = self._head
self._size = 0
[документация]
def index(self, item: T, start: int = 0, end: int | None = None) -> int:
"""
Возвращает индекс первого вхождения элемента в заданном диапазоне.
:param item: Искомый элемент.
:param start: Начальный индекс поиска.
:param end: Конечный индекс поиска (не включая его).
:return: Порядковый индекс элемента.
:raises ValueError: Если элемент не найден в указанных границах.
"""
with self._lock:
if start < 0:
start = max(0, start + self._size)
if end is None:
end = self._size
elif end < 0:
end = max(0, end + self._size)
current = self._head.next
idx = 0
while current is not self._tail:
if start <= idx < end and current.value == item: # type: ignore
return idx
current = current.next # type: ignore
idx += 1
raise ValueError(f"'{item}' is not in list")
[документация]
def count(self, item: T) -> int:
"""
Возвращает количество вхождений элемента в список.
:param item: Элемент для подсчета.
:return: Количество совпадений.
"""
with self._lock:
cnt = 0
current = self._head.next
while current is not self._tail:
if current.value == item: # type: ignore
cnt += 1
current = current.next # type: ignore
return cnt
[документация]
def reverse(self) -> None:
"""Разворачивает список на месте."""
with self._lock:
current = self._head
while current is not None:
current.prev, current.next = current.next, current.prev
current = current.prev # type: ignore
self._head, self._tail = self._tail, self._head
[документация]
def sort(self, key: Callable[[T], T] | None = None, reverse: bool = False) -> None:
"""
Сортирует элементы списка на месте.
:param key: Функция, извлекающая ключ для сравнения из каждого элемента.
:param reverse: Если True, сортировка производится по убыванию.
:return: None
"""
with self._lock:
py_list: list[T] = []
current = self._head.next
while current is not self._tail:
py_list.append(current.value) # type: ignore
current = current.next # type: ignore
py_list.sort(key=key, reverse=reverse) # type: ignore
current = self._head.next
for val in py_list:
current.value = val # type: ignore
current = current.next # type: ignore
@overload
def __getitem__(self, index: int) -> T: ...
@overload
def __getitem__(self, index: slice) -> Self: ...
def __getitem__(self, index: int | slice) -> T | Self:
"""
Возвращает элемент по индексу или новый список (срез).
:param index: Целочисленный индекс или объект slice.
:return: Элемент типа T или новый объект List[T] для среза.
:raises IndexError: Если индекс вне диапазона.
"""
with self._lock:
if isinstance(index, slice):
start, stop, step = index.indices(self._size)
new_list = self.__class__()
py_list: list[T] = []
current = self._head.next
while current is not self._tail:
py_list.append(current.value) # type: ignore
current = current.next # type: ignore
new_list.extend(py_list[start:stop:step])
return new_list
else:
return self._get_node(index).value # type: ignore
@overload
def __setitem__(self, index: int, value: T) -> None: ...
@overload
def __setitem__(self, index: slice, value: Iterable[T]) -> None: ...
def __setitem__(self, index: int | slice, value: T | Iterable[T]) -> None:
"""
Устанавливает новое значение элементу или срезу.
:param index: Целочисленный индекс или срез.
:param value: Новое значение типа T или итерируемый объект для среза.
"""
with self._lock:
if isinstance(index, slice):
py_list: list[T] = []
current = self._head.next
while current is not self._tail:
py_list.append(current.value) # type: ignore
current = current.next # type: ignore
py_list[index] = value # type: ignore
self.clear()
self.extend(py_list)
else:
self._get_node(index).value = value # type: ignore
@overload
def __delitem__(self, index: int) -> None: ...
@overload
def __delitem__(self, index: slice) -> None: ...
def __delitem__(self, index: int | slice) -> None:
"""
Удаляет элемент или срез по индексу.
:param index: Целочисленный индекс или срез.
"""
with self._lock:
if isinstance(index, slice):
py_list: list[T] = []
current = self._head.next
while current is not self._tail:
py_list.append(current.value) # type: ignore
current = current.next # type: ignore
del py_list[index]
self.clear()
self.extend(py_list)
else:
node = self._get_node(index)
node.prev.next = node.next # type: ignore
node.next.prev = node.prev # type: ignore
self._size -= 1
def __add__(self, other: Iterable[T]) -> Self:
"""
Оператор сложения (self + other).
Создает новый список.
:param other: Итерируемый объект с элементами типа T.
:return: Новый экземпляр класса, объединяющий оба списка.
"""
if not isinstance(other, Iterable):
return NotImplemented
new_list = self.__class__()
new_list.extend(self)
new_list.extend(other)
return new_list
def __radd__(self, other: Iterable[T]) -> Self:
"""
Оператор правого сложения (other + self).
Создает новый список.
:param other: Итерируемый объект с элементами типа T.
:return: Новый экземпляр класса, объединяющий оба списка.
"""
if not isinstance(other, Iterable):
return NotImplemented
new_list = self.__class__()
new_list.extend(other)
new_list.extend(self)
return new_list
def __iadd__(self, other: Iterable[T]) -> Self:
"""
Оператор сложения на месте (self += other.
:param other: Итерируемый объект с элементами типа T.
:return: Текущий измененный объект списка.
"""
if not isinstance(other, Iterable):
return NotImplemented
self.extend(other)
return self
def __mul__(self, count: int) -> Self:
"""
Оператор умножения списка на число (self * count).
:param count: Количество повторений.
:return: Новый дублированный список.
"""
if not isinstance(count, int):
return NotImplemented
new_list = self.__class__()
if count <= 0:
return new_list
for _ in range(count):
new_list.extend(self)
return new_list
def __rmul__(self, count: int) -> Self:
"""
Оператор правого умножения списка на число (count * self).
:param count: Количество повторений.
:return: Новый дублированный список.
"""
return self.__mul__(count)
def __imul__(self, count: int) -> Self:
"""
Оператор умножения на месте (self *= count).
:param count: Количество повторений.
:return: Текущий измененный объект списка.
"""
if not isinstance(count, int):
return NotImplemented
if count <= 0:
self.clear()
return self
snapshot = list(self)
for _ in range(count - 1):
self.extend(snapshot)
return self
def __iter__(self) -> ListIterator[T]:
"""
Возвращает специальный итератор, устойчивый к изменениям.
:return: Экземпляр ListIterator[T].
"""
return ListIterator(self._head, self._tail, self._lock)
def __len__(self) -> int:
"""
Возвращает текущую длину (размер) списка.
:return: Количество элементов в списке.
"""
with self._lock:
return self._size
def __contains__(self, item: T) -> bool:
"""
Проверяет наличие элемента в списке оператором in.
:param item: Искомый элемент.
:return: True, если элемент найден, иначе False.
"""
with self._lock:
current = self._head.next
while current is not self._tail:
if current.value == item: # type: ignore
return True
current = current.next # type: ignore
return False
def __str__(self) -> str:
"""
Возвращает строковое представление списка.
:return: Представление списка в виде строки.
"""
with self._lock:
elements = []
current = self._head.next
while current is not self._tail:
elements.append(repr(current.value)) # type: ignore
current = current.next # type: ignore
return f"[{', '.join(elements)}]"
def __repr__(self) -> str:
"""
Возвращает строковое представление списка.
:return: Представление списка в виде строки.
"""
return f"List({str(self)})"
def __del__(self) -> None:
"""Очищает память."""
self.clear()