Исходный код EJIO.utils.struct.deque

## /utils/struct/deque.py

from threading import Lock
from typing import Any, Generator

__all__ = [
    'Deque',
]


class Node[T]:
    """
    Узел двусвязного списка для хранения значения и указателей на следующий и предыдущий элементы.
    """
    __slots__ = ('value', 'next', 'prev')

    def __init__(
            self,
            value: T,
            next: 'Node[T] | None' = None,
            prev: 'Node[T] | None' = None
    ) -> None:
        """
        Создает узел двусвязного списка.

        :param value: Значение записываемое в узел.
        :param next: Ссылка на следующий узел.
        :param prev: Ссылка на предыдущий узел.
        """
        self.value: T = value
        self.next: 'Node[T] | None' = next
        self.prev: 'Node[T] | None' = prev




class Deque[T]:
    """
    Потокобезопасная двусторонняя очередь на двусвязном списке.
    Класс поддерживает основные операции с обоих концов структуры.
    """
    __slots__ = ('_head', '_tail', '_size', '_lock')

    def __init__(self) -> None:
        """Создает объект двусторонней очереди."""
        self._head: Node[T] | None = None
        self._tail: Node[T] | None = None
        self._lock: Lock = Lock()
        self._size: int = 0



    def push_back(self, item: T) -> None:
        """
        Добавляет элемент в конец (хвост) очереди.

        :param item: Объект для добавления.
        """
        with self._lock:
            new_node = Node(item, prev=self._tail)
            if self._tail is None:
                self._head = new_node
                self._tail = new_node
            else:
                self._tail.next = new_node
                self._tail = new_node
            self._size += 1




    def push_front(self, item: T) -> None:
        """
        Добавляет элемент в начало (голову) очереди.

        :param item: Объект для добавления.
        """
        with self._lock:
            new_node = Node(item, next=self._head)
            if self._head is None:
                self._head = new_node
                self._tail = new_node
            else:
                self._head.prev = new_node
                self._head = new_node
            self._size += 1




    def pop_front(self) -> T:
        """
        Удаляет и возвращает элемент из начала (головы) очереди.

        :return: Объект из начала очереди.
        :raises IndexError: Если очередь пуста.
        """
        with self._lock:
            if self._head is None:
                raise IndexError("Pop from empty deque")
            node = self._head
            self._head = node.next

            if self._head is None:
                self._tail = None
            else:
                self._head.prev = None

            self._size -= 1
            return node.value




    def pop_back(self) -> T:
        """
        Удаляет и возвращает элемент из конца (хвоста) очереди.

        :return: Объект из конца очереди.
        :raises IndexError: Если очередь пуста.
        """
        with self._lock:
            if self._tail is None:
                raise IndexError("Pop from empty deque")
            node = self._tail
            self._tail = node.prev

            if self._tail is None:
                self._head = None
            else:
                self._tail.next = None

            self._size -= 1
            return node.value




    def peek_front(self) -> T:
        """
        Возвращает первый элемент (в голове) без удаления.

        :return: Объект из начала очереди.
        :raises IndexError: Если очередь пуста.
        """
        with self._lock:
            if self._head is None:
                raise IndexError("Peek from empty deque")
            return self._head.value




    def peek_back(self) -> T:
        """
        Возвращает последний элемент (в хвосте) без удаления.

        :return: Объект из конца очереди.
        :raises IndexError: Если очередь пуста.
        """
        with self._lock:
            if self._tail is None:
                raise IndexError("Peek from empty deque")
            return self._tail.value


    @property
    def empty(self) -> bool:
        """
        Проверяет пуста ли очередь.

        :return: True если очередь пуста, иначе False.
        """
        with self._lock:
            return self._head is None

    @property
    def size(self) -> int:
        """
        Возвращает текущий размер очереди.

        :return: Размер очереди.
        """
        with self._lock:
            return self._size

    def __reversed__(self) -> Generator[T, Any, None]:
        """
        Возвращает итератор, извлекающий элементы
        из конца очереди до тех пор, пока она не опустеет.
        Потокобезопасно на уровне отдельных операций.

        :yields: Элементы очереди в порядке с конца до начала.
        """
        while True:
            try:
                yield self.pop_back()
            except IndexError:
                return

    def __iter__(self) -> Generator[T, Any, None]:
        """
        Возвращает итератор, извлекающий элементы
        из начала очереди до тех пор, пока она не опустеет.
        Потокобезопасно на уровне отдельных операций.

        :yields: Элементы очереди в порядке с начала до конца.
        """
        while True:
            try:
                yield self.pop_front()
            except IndexError:
                return

    def __str__(self) -> str:
        """
        Возвращает строковое представление всей очереди в виде списка элементов
        от начала к концу.

        :return: Представление очереди в виде строки.
        """
        with self._lock:
            elements = []
            current = self._head
            while current:
                elements.append(repr(current.value))
                current = current.next
            return f"[{', '.join(elements)}]"

    def __repr__(self) -> str:
        """
        Возвращает строковое представление всего дека в виде списка элементов
        от начала к концу.

        :return: Представление очереди в виде строки.
        """
        return f"Deque({str(self)})"

    def __del__(self) -> None:
        """Очищает память при удалении объекта, изолируя узлы двусвязного списка."""
        with self._lock:
            current = self._head
            while current:
                next_node = current.next
                current.next = None
                current.prev = None
                current = next_node
            self._head = None
            self._tail = None
            self._size = 0