algorithm/Python
1
0
Fork 0
mirror of https://github.com/TheAlgorithms/Python.git synced 2025-07-06 18:49:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

git mv data_structures/queue data_structures/queues (#12577)

Browse Source 
Co-authored-by: Christian Clauss <cclauss@me.com>
This commit is contained in:
pre-commit-ci[bot]
2025-02-14 10:05:23 +01:00
committed by GitHub
parent 338cbafe0d
commit 738253e800
11 changed files with 21 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
data_structures/queues/__init__.py Normal file
View File

101
data_structures/queues/circular_queue.py Normal file
View File

@ -0,0 +1,101 @@
# Implementation of Circular Queue (using Python lists)
class CircularQueue:
"""Circular FIFO queue with a fixed capacity"""
def __init__(self, n: int):
self.n = n
self.array = [None] * self.n
self.front = 0 # index of the first element
self.rear = 0
self.size = 0
def __len__(self) -> int:
"""
>>> cq = CircularQueue(5)
>>> len(cq)
0
>>> cq.enqueue("A") # doctest: +ELLIPSIS
<data_structures.queues.circular_queue.CircularQueue object at ...
>>> cq.array
['A', None, None, None, None]
>>> len(cq)
1
"""
return self.size
def is_empty(self) -> bool:
"""
Checks whether the queue is empty or not
>>> cq = CircularQueue(5)
>>> cq.is_empty()
True
>>> cq.enqueue("A").is_empty()
False
"""
return self.size == 0
def first(self):
"""
Returns the first element of the queue
>>> cq = CircularQueue(5)
>>> cq.first()
False
>>> cq.enqueue("A").first()
'A'
"""
return False if self.is_empty() else self.array[self.front]
def enqueue(self, data):
"""
This function inserts an element at the end of the queue using self.rear value
as an index.
>>> cq = CircularQueue(5)
>>> cq.enqueue("A") # doctest: +ELLIPSIS
<data_structures.queues.circular_queue.CircularQueue object at ...
>>> (cq.size, cq.first())
(1, 'A')
>>> cq.enqueue("B") # doctest: +ELLIPSIS
<data_structures.queues.circular_queue.CircularQueue object at ...
>>> cq.array
['A', 'B', None, None, None]
>>> (cq.size, cq.first())
(2, 'A')
"""
if self.size >= self.n:
raise Exception("QUEUE IS FULL")
self.array[self.rear] = data
self.rear = (self.rear + 1) % self.n
self.size += 1
return self
def dequeue(self):
"""
This function removes an element from the queue using on self.front value as an
index and returns it
>>> cq = CircularQueue(5)
>>> cq.dequeue()
Traceback (most recent call last):
...
Exception: UNDERFLOW
>>> cq.enqueue("A").enqueue("B").dequeue()
'A'
>>> (cq.size, cq.first())
(1, 'B')
>>> cq.dequeue()
'B'
>>> cq.dequeue()
Traceback (most recent call last):
...
Exception: UNDERFLOW
"""
if self.size == 0:
raise Exception("UNDERFLOW")
temp = self.array[self.front]
self.array[self.front] = None
self.front = (self.front + 1) % self.n
self.size -= 1
return temp

161
data_structures/queues/circular_queue_linked_list.py Normal file
View File

@ -0,0 +1,161 @@
# Implementation of Circular Queue using linked lists
# https://en.wikipedia.org/wiki/Circular_buffer
from __future__ import annotations
from typing import Any
class CircularQueueLinkedList:
"""
Circular FIFO list with the given capacity (default queue length : 6)
>>> cq = CircularQueueLinkedList(2)
>>> cq.enqueue('a')
>>> cq.enqueue('b')
>>> cq.enqueue('c')
Traceback (most recent call last):
...
Exception: Full Queue
"""
def __init__(self, initial_capacity: int = 6) -> None:
self.front: Node | None = None
self.rear: Node | None = None
self.create_linked_list(initial_capacity)
def create_linked_list(self, initial_capacity: int) -> None:
current_node = Node()
self.front = current_node
self.rear = current_node
previous_node = current_node
for _ in range(1, initial_capacity):
current_node = Node()
previous_node.next = current_node
current_node.prev = previous_node
previous_node = current_node
previous_node.next = self.front
self.front.prev = previous_node
def is_empty(self) -> bool:
"""
Checks whether the queue is empty or not
>>> cq = CircularQueueLinkedList()
>>> cq.is_empty()
True
>>> cq.enqueue('a')
>>> cq.is_empty()
False
>>> cq.dequeue()
'a'
>>> cq.is_empty()
True
"""
return (
self.front == self.rear
and self.front is not None
and self.front.data is None
)
def first(self) -> Any | None:
"""
Returns the first element of the queue
>>> cq = CircularQueueLinkedList()
>>> cq.first()
Traceback (most recent call last):
...
Exception: Empty Queue
>>> cq.enqueue('a')
>>> cq.first()
'a'
>>> cq.dequeue()
'a'
>>> cq.first()
Traceback (most recent call last):
...
Exception: Empty Queue
>>> cq.enqueue('b')
>>> cq.enqueue('c')
>>> cq.first()
'b'
"""
self.check_can_perform_operation()
return self.front.data if self.front else None
def enqueue(self, data: Any) -> None:
"""
Saves data at the end of the queue
>>> cq = CircularQueueLinkedList()
>>> cq.enqueue('a')
>>> cq.enqueue('b')
>>> cq.dequeue()
'a'
>>> cq.dequeue()
'b'
>>> cq.dequeue()
Traceback (most recent call last):
...
Exception: Empty Queue
"""
if self.rear is None:
return
self.check_is_full()
if not self.is_empty():
self.rear = self.rear.next
if self.rear:
self.rear.data = data
def dequeue(self) -> Any:
"""
Removes and retrieves the first element of the queue
>>> cq = CircularQueueLinkedList()
>>> cq.dequeue()
Traceback (most recent call last):
...
Exception: Empty Queue
>>> cq.enqueue('a')
>>> cq.dequeue()
'a'
>>> cq.dequeue()
Traceback (most recent call last):
...
Exception: Empty Queue
"""
self.check_can_perform_operation()
if self.rear is None or self.front is None:
return None
if self.front == self.rear:
data = self.front.data
self.front.data = None
return data
old_front = self.front
self.front = old_front.next
data = old_front.data
old_front.data = None
return data
def check_can_perform_operation(self) -> None:
if self.is_empty():
raise Exception("Empty Queue")
def check_is_full(self) -> None:
if self.rear and self.rear.next == self.front:
raise Exception("Full Queue")
class Node:
def __init__(self) -> None:
self.data: Any | None = None
self.next: Node | None = None
self.prev: Node | None = None
if __name__ == "__main__":
import doctest
doctest.testmod()

463
data_structures/queues/double_ended_queue.py Normal file
View File

@ -0,0 +1,463 @@
"""
Implementation of double ended queue.
"""
from __future__ import annotations
from collections.abc import Iterable
from dataclasses import dataclass
from typing import Any
class Deque:
"""
Deque data structure.
Operations
----------
append(val: Any) -> None
appendleft(val: Any) -> None
extend(iterable: Iterable) -> None
extendleft(iterable: Iterable) -> None
pop() -> Any
popleft() -> Any
Observers
---------
is_empty() -> bool
Attributes
----------
_front: _Node
front of the deque a.k.a. the first element
_back: _Node
back of the element a.k.a. the last element
_len: int
the number of nodes
"""
__slots__ = ("_back", "_front", "_len")
@dataclass
class _Node:
"""
Representation of a node.
Contains a value and a pointer to the next node as well as to the previous one.
"""
val: Any = None
next_node: Deque._Node | None = None
prev_node: Deque._Node | None = None
class _Iterator:
"""
Helper class for iteration. Will be used to implement iteration.
Attributes
----------
_cur: _Node
the current node of the iteration.
"""
__slots__ = ("_cur",)
def __init__(self, cur: Deque._Node | None) -> None:
self._cur = cur
def __iter__(self) -> Deque._Iterator:
"""
>>> our_deque = Deque([1, 2, 3])
>>> iterator = iter(our_deque)
"""
return self
def __next__(self) -> Any:
"""
>>> our_deque = Deque([1, 2, 3])
>>> iterator = iter(our_deque)
>>> next(iterator)
1
>>> next(iterator)
2
>>> next(iterator)
3
"""
if self._cur is None:
# finished iterating
raise StopIteration
val = self._cur.val
self._cur = self._cur.next_node
return val
def __init__(self, iterable: Iterable[Any] | None = None) -> None:
self._front: Any = None
self._back: Any = None
self._len: int = 0
if iterable is not None:
# append every value to the deque
for val in iterable:
self.append(val)
def append(self, val: Any) -> None:
"""
Adds val to the end of the deque.
Time complexity: O(1)
>>> our_deque_1 = Deque([1, 2, 3])
>>> our_deque_1.append(4)
>>> our_deque_1
[1, 2, 3, 4]
>>> our_deque_2 = Deque('ab')
>>> our_deque_2.append('c')
>>> our_deque_2
['a', 'b', 'c']
>>> from collections import deque
>>> deque_collections_1 = deque([1, 2, 3])
>>> deque_collections_1.append(4)
>>> deque_collections_1
deque([1, 2, 3, 4])
>>> deque_collections_2 = deque('ab')
>>> deque_collections_2.append('c')
>>> deque_collections_2
deque(['a', 'b', 'c'])
>>> list(our_deque_1) == list(deque_collections_1)
True
>>> list(our_deque_2) == list(deque_collections_2)
True
"""
node = self._Node(val, None, None)
if self.is_empty():
# front = back
self._front = self._back = node
self._len = 1
else:
# connect nodes
self._back.next_node = node
node.prev_node = self._back
self._back = node # assign new back to the new node
self._len += 1
# make sure there were no errors
assert not self.is_empty(), "Error on appending value."
def appendleft(self, val: Any) -> None:
"""
Adds val to the beginning of the deque.
Time complexity: O(1)
>>> our_deque_1 = Deque([2, 3])
>>> our_deque_1.appendleft(1)
>>> our_deque_1
[1, 2, 3]
>>> our_deque_2 = Deque('bc')
>>> our_deque_2.appendleft('a')
>>> our_deque_2
['a', 'b', 'c']
>>> from collections import deque
>>> deque_collections_1 = deque([2, 3])
>>> deque_collections_1.appendleft(1)
>>> deque_collections_1
deque([1, 2, 3])
>>> deque_collections_2 = deque('bc')
>>> deque_collections_2.appendleft('a')
>>> deque_collections_2
deque(['a', 'b', 'c'])
>>> list(our_deque_1) == list(deque_collections_1)
True
>>> list(our_deque_2) == list(deque_collections_2)
True
"""
node = self._Node(val, None, None)
if self.is_empty():
# front = back
self._front = self._back = node
self._len = 1
else:
# connect nodes
node.next_node = self._front
self._front.prev_node = node
self._front = node # assign new front to the new node
self._len += 1
# make sure there were no errors
assert not self.is_empty(), "Error on appending value."
def extend(self, iterable: Iterable[Any]) -> None:
"""
Appends every value of iterable to the end of the deque.
Time complexity: O(n)
>>> our_deque_1 = Deque([1, 2, 3])
>>> our_deque_1.extend([4, 5])
>>> our_deque_1
[1, 2, 3, 4, 5]
>>> our_deque_2 = Deque('ab')
>>> our_deque_2.extend('cd')
>>> our_deque_2
['a', 'b', 'c', 'd']
>>> from collections import deque
>>> deque_collections_1 = deque([1, 2, 3])
>>> deque_collections_1.extend([4, 5])
>>> deque_collections_1
deque([1, 2, 3, 4, 5])
>>> deque_collections_2 = deque('ab')
>>> deque_collections_2.extend('cd')
>>> deque_collections_2
deque(['a', 'b', 'c', 'd'])
>>> list(our_deque_1) == list(deque_collections_1)
True
>>> list(our_deque_2) == list(deque_collections_2)
True
"""
for val in iterable:
self.append(val)
def extendleft(self, iterable: Iterable[Any]) -> None:
"""
Appends every value of iterable to the beginning of the deque.
Time complexity: O(n)
>>> our_deque_1 = Deque([1, 2, 3])
>>> our_deque_1.extendleft([0, -1])
>>> our_deque_1
[-1, 0, 1, 2, 3]
>>> our_deque_2 = Deque('cd')
>>> our_deque_2.extendleft('ba')
>>> our_deque_2
['a', 'b', 'c', 'd']
>>> from collections import deque
>>> deque_collections_1 = deque([1, 2, 3])
>>> deque_collections_1.extendleft([0, -1])
>>> deque_collections_1
deque([-1, 0, 1, 2, 3])
>>> deque_collections_2 = deque('cd')
>>> deque_collections_2.extendleft('ba')
>>> deque_collections_2
deque(['a', 'b', 'c', 'd'])
>>> list(our_deque_1) == list(deque_collections_1)
True
>>> list(our_deque_2) == list(deque_collections_2)
True
"""
for val in iterable:
self.appendleft(val)
def pop(self) -> Any:
"""
Removes the last element of the deque and returns it.
Time complexity: O(1)
@returns topop.val: the value of the node to pop.
>>> our_deque1 = Deque([1])
>>> our_popped1 = our_deque1.pop()
>>> our_popped1
1
>>> our_deque1
[]
>>> our_deque2 = Deque([1, 2, 3, 15182])
>>> our_popped2 = our_deque2.pop()
>>> our_popped2
15182
>>> our_deque2
[1, 2, 3]
>>> from collections import deque
>>> deque_collections = deque([1, 2, 3, 15182])
>>> collections_popped = deque_collections.pop()
>>> collections_popped
15182
>>> deque_collections
deque([1, 2, 3])
>>> list(our_deque2) == list(deque_collections)
True
>>> our_popped2 == collections_popped
True
"""
# make sure the deque has elements to pop
assert not self.is_empty(), "Deque is empty."
topop = self._back
# if only one element in the queue: point the front and back to None
# else remove one element from back
if self._front == self._back:
self._front = None
self._back = None
else:
self._back = self._back.prev_node # set new back
# drop the last node, python will deallocate memory automatically
self._back.next_node = None
self._len -= 1
return topop.val
def popleft(self) -> Any:
"""
Removes the first element of the deque and returns it.
Time complexity: O(1)
@returns topop.val: the value of the node to pop.
>>> our_deque1 = Deque([1])
>>> our_popped1 = our_deque1.pop()
>>> our_popped1
1
>>> our_deque1
[]
>>> our_deque2 = Deque([15182, 1, 2, 3])
>>> our_popped2 = our_deque2.popleft()
>>> our_popped2
15182
>>> our_deque2
[1, 2, 3]
>>> from collections import deque
>>> deque_collections = deque([15182, 1, 2, 3])
>>> collections_popped = deque_collections.popleft()
>>> collections_popped
15182
>>> deque_collections
deque([1, 2, 3])
>>> list(our_deque2) == list(deque_collections)
True
>>> our_popped2 == collections_popped
True
"""
# make sure the deque has elements to pop
assert not self.is_empty(), "Deque is empty."
topop = self._front
# if only one element in the queue: point the front and back to None
# else remove one element from front
if self._front == self._back:
self._front = None
self._back = None
else:
self._front = self._front.next_node # set new front and drop the first node
self._front.prev_node = None
self._len -= 1
return topop.val
def is_empty(self) -> bool:
"""
Checks if the deque is empty.
Time complexity: O(1)
>>> our_deque = Deque([1, 2, 3])
>>> our_deque.is_empty()
False
>>> our_empty_deque = Deque()
>>> our_empty_deque.is_empty()
True
>>> from collections import deque
>>> empty_deque_collections = deque()
>>> list(our_empty_deque) == list(empty_deque_collections)
True
"""
return self._front is None
def __len__(self) -> int:
"""
Implements len() function. Returns the length of the deque.
Time complexity: O(1)
>>> our_deque = Deque([1, 2, 3])
>>> len(our_deque)
3
>>> our_empty_deque = Deque()
>>> len(our_empty_deque)
0
>>> from collections import deque
>>> deque_collections = deque([1, 2, 3])
>>> len(deque_collections)
3
>>> empty_deque_collections = deque()
>>> len(empty_deque_collections)
0
>>> len(our_empty_deque) == len(empty_deque_collections)
True
"""
return self._len
def __eq__(self, other: object) -> bool:
"""
Implements "==" operator. Returns if *self* is equal to *other*.
Time complexity: O(n)
>>> our_deque_1 = Deque([1, 2, 3])
>>> our_deque_2 = Deque([1, 2, 3])
>>> our_deque_1 == our_deque_2
True
>>> our_deque_3 = Deque([1, 2])
>>> our_deque_1 == our_deque_3
False
>>> from collections import deque
>>> deque_collections_1 = deque([1, 2, 3])
>>> deque_collections_2 = deque([1, 2, 3])
>>> deque_collections_1 == deque_collections_2
True
>>> deque_collections_3 = deque([1, 2])
>>> deque_collections_1 == deque_collections_3
False
>>> (our_deque_1 == our_deque_2) == (deque_collections_1 == deque_collections_2)
True
>>> (our_deque_1 == our_deque_3) == (deque_collections_1 == deque_collections_3)
True
"""
if not isinstance(other, Deque):
return NotImplemented
me = self._front
oth = other._front
# if the length of the dequeues are not the same, they are not equal
if len(self) != len(other):
return False
while me is not None and oth is not None:
# compare every value
if me.val != oth.val:
return False
me = me.next_node
oth = oth.next_node
return True
def __iter__(self) -> Deque._Iterator:
"""
Implements iteration.
Time complexity: O(1)
>>> our_deque = Deque([1, 2, 3])
>>> for v in our_deque:
... print(v)
1
2
3
>>> from collections import deque
>>> deque_collections = deque([1, 2, 3])
>>> for v in deque_collections:
... print(v)
1
2
3
"""
return Deque._Iterator(self._front)
def __repr__(self) -> str:
"""
Implements representation of the deque.
Represents it as a list, with its values between '[' and ']'.
Time complexity: O(n)
>>> our_deque = Deque([1, 2, 3])
>>> our_deque
[1, 2, 3]
"""
values_list = []
aux = self._front
while aux is not None:
# append the values in a list to display
values_list.append(aux.val)
aux = aux.next_node
return f"[{', '.join(repr(val) for val in values_list)}]"
if __name__ == "__main__":
import doctest
doctest.testmod()
dq = Deque([3])
dq.pop()

156
data_structures/queues/linked_queue.py Normal file
View File

@ -0,0 +1,156 @@
"""A Queue using a linked list like structure"""
from __future__ import annotations
from collections.abc import Iterator
from typing import Any
class Node:
def __init__(self, data: Any) -> None:
self.data: Any = data
self.next: Node | None = None
def __str__(self) -> str:
return f"{self.data}"
class LinkedQueue:
"""
>>> queue = LinkedQueue()
>>> queue.is_empty()
True
>>> queue.put(5)
>>> queue.put(9)
>>> queue.put('python')
>>> queue.is_empty()
False
>>> queue.get()
5
>>> queue.put('algorithms')
>>> queue.get()
9
>>> queue.get()
'python'
>>> queue.get()
'algorithms'
>>> queue.is_empty()
True
>>> queue.get()
Traceback (most recent call last):
...
IndexError: dequeue from empty queue
"""
def __init__(self) -> None:
self.front: Node | None = None
self.rear: Node | None = None
def __iter__(self) -> Iterator[Any]:
node = self.front
while node:
yield node.data
node = node.next
def __len__(self) -> int:
"""
>>> queue = LinkedQueue()
>>> for i in range(1, 6):
... queue.put(i)
>>> len(queue)
5
>>> for i in range(1, 6):
... assert len(queue) == 6 - i
... _ = queue.get()
>>> len(queue)
0
"""
return len(tuple(iter(self)))
def __str__(self) -> str:
"""
>>> queue = LinkedQueue()
>>> for i in range(1, 4):
... queue.put(i)
>>> queue.put("Python")
>>> queue.put(3.14)
>>> queue.put(True)
>>> str(queue)
'1 <- 2 <- 3 <- Python <- 3.14 <- True'
"""
return " <- ".join(str(item) for item in self)
def is_empty(self) -> bool:
"""
>>> queue = LinkedQueue()
>>> queue.is_empty()
True
>>> for i in range(1, 6):
... queue.put(i)
>>> queue.is_empty()
False
"""
return len(self) == 0
def put(self, item: Any) -> None:
"""
>>> queue = LinkedQueue()
>>> queue.get()
Traceback (most recent call last):
...
IndexError: dequeue from empty queue
>>> for i in range(1, 6):
... queue.put(i)
>>> str(queue)
'1 <- 2 <- 3 <- 4 <- 5'
"""
node = Node(item)
if self.is_empty():
self.front = self.rear = node
else:
assert isinstance(self.rear, Node)
self.rear.next = node
self.rear = node
def get(self) -> Any:
"""
>>> queue = LinkedQueue()
>>> queue.get()
Traceback (most recent call last):
...
IndexError: dequeue from empty queue
>>> queue = LinkedQueue()
>>> for i in range(1, 6):
... queue.put(i)
>>> for i in range(1, 6):
... assert queue.get() == i
>>> len(queue)
0
"""
if self.is_empty():
raise IndexError("dequeue from empty queue")
assert isinstance(self.front, Node)
node = self.front
self.front = self.front.next
if self.front is None:
self.rear = None
return node.data
def clear(self) -> None:
"""
>>> queue = LinkedQueue()
>>> for i in range(1, 6):
... queue.put(i)
>>> queue.clear()
>>> len(queue)
0
>>> str(queue)
''
"""
self.front = self.rear = None
if __name__ == "__main__":
from doctest import testmod
testmod()

232
data_structures/queues/priority_queue_using_list.py Normal file
View File

@ -0,0 +1,232 @@
"""
Pure Python implementations of a Fixed Priority Queue and an Element Priority Queue
using Python lists.
"""
class OverFlowError(Exception):
pass
class UnderFlowError(Exception):
pass
class FixedPriorityQueue:
"""
Tasks can be added to a Priority Queue at any time and in any order but when Tasks
are removed then the Task with the highest priority is removed in FIFO order. In
code we will use three levels of priority with priority zero Tasks being the most
urgent (high priority) and priority 2 tasks being the least urgent.
Examples
>>> fpq = FixedPriorityQueue()
>>> fpq.enqueue(0, 10)
>>> fpq.enqueue(1, 70)
>>> fpq.enqueue(0, 100)
>>> fpq.enqueue(2, 1)
>>> fpq.enqueue(2, 5)
>>> fpq.enqueue(1, 7)
>>> fpq.enqueue(2, 4)
>>> fpq.enqueue(1, 64)
>>> fpq.enqueue(0, 128)
>>> print(fpq)
Priority 0: [10, 100, 128]
Priority 1: [70, 7, 64]
Priority 2: [1, 5, 4]
>>> fpq.dequeue()
10
>>> fpq.dequeue()
100
>>> fpq.dequeue()
128
>>> fpq.dequeue()
70
>>> fpq.dequeue()
7
>>> print(fpq)
Priority 0: []
Priority 1: [64]
Priority 2: [1, 5, 4]
>>> fpq.dequeue()
64
>>> fpq.dequeue()
1
>>> fpq.dequeue()
5
>>> fpq.dequeue()
4
>>> fpq.dequeue()
Traceback (most recent call last):
...
data_structures.queues.priority_queue_using_list.UnderFlowError: All queues are empty
>>> print(fpq)
Priority 0: []
Priority 1: []
Priority 2: []
""" # noqa: E501
def __init__(self):
self.queues = [
[],
[],
[],
]
def enqueue(self, priority: int, data: int) -> None:
"""
Add an element to a queue based on its priority.
If the priority is invalid ValueError is raised.
If the queue is full an OverFlowError is raised.
"""
try:
if len(self.queues[priority]) >= 100:
raise OverflowError("Maximum queue size is 100")
self.queues[priority].append(data)
except IndexError:
raise ValueError("Valid priorities are 0, 1, and 2")
def dequeue(self) -> int:
"""
Return the highest priority element in FIFO order.
If the queue is empty then an under flow exception is raised.
"""
for queue in self.queues:
if queue:
return queue.pop(0)
raise UnderFlowError("All queues are empty")
def __str__(self) -> str:
return "\n".join(f"Priority {i}: {q}" for i, q in enumerate(self.queues))
class ElementPriorityQueue:
"""
Element Priority Queue is the same as Fixed Priority Queue except that the value of
the element itself is the priority. The rules for priorities are the same the as
Fixed Priority Queue.
>>> epq = ElementPriorityQueue()
>>> epq.enqueue(10)
>>> epq.enqueue(70)
>>> epq.enqueue(4)
>>> epq.enqueue(1)
>>> epq.enqueue(5)
>>> epq.enqueue(7)
>>> epq.enqueue(4)
>>> epq.enqueue(64)
>>> epq.enqueue(128)
>>> print(epq)
[10, 70, 4, 1, 5, 7, 4, 64, 128]
>>> epq.dequeue()
1
>>> epq.dequeue()
4
>>> epq.dequeue()
4
>>> epq.dequeue()
5
>>> epq.dequeue()
7
>>> epq.dequeue()
10
>>> print(epq)
[70, 64, 128]
>>> epq.dequeue()
64
>>> epq.dequeue()
70
>>> epq.dequeue()
128
>>> epq.dequeue()
Traceback (most recent call last):
...
data_structures.queues.priority_queue_using_list.UnderFlowError: The queue is empty
>>> print(epq)
[]
"""
def __init__(self):
self.queue = []
def enqueue(self, data: int) -> None:
"""
This function enters the element into the queue
If the queue is full an Exception is raised saying Over Flow!
"""
if len(self.queue) == 100:
raise OverFlowError("Maximum queue size is 100")
self.queue.append(data)
def dequeue(self) -> int:
"""
Return the highest priority element in FIFO order.
If the queue is empty then an under flow exception is raised.
"""
if not self.queue:
raise UnderFlowError("The queue is empty")
else:
data = min(self.queue)
self.queue.remove(data)
return data
def __str__(self) -> str:
"""
Prints all the elements within the Element Priority Queue
"""
return str(self.queue)
def fixed_priority_queue():
fpq = FixedPriorityQueue()
fpq.enqueue(0, 10)
fpq.enqueue(1, 70)
fpq.enqueue(0, 100)
fpq.enqueue(2, 1)
fpq.enqueue(2, 5)
fpq.enqueue(1, 7)
fpq.enqueue(2, 4)
fpq.enqueue(1, 64)
fpq.enqueue(0, 128)
print(fpq)
print(fpq.dequeue())
print(fpq.dequeue())
print(fpq.dequeue())
print(fpq.dequeue())
print(fpq.dequeue())
print(fpq)
print(fpq.dequeue())
print(fpq.dequeue())
print(fpq.dequeue())
print(fpq.dequeue())
print(fpq.dequeue())
def element_priority_queue():
epq = ElementPriorityQueue()
epq.enqueue(10)
epq.enqueue(70)
epq.enqueue(100)
epq.enqueue(1)
epq.enqueue(5)
epq.enqueue(7)
epq.enqueue(4)
epq.enqueue(64)
epq.enqueue(128)
print(epq)
print(epq.dequeue())
print(epq.dequeue())
print(epq.dequeue())
print(epq.dequeue())
print(epq.dequeue())
print(epq)
print(epq.dequeue())
print(epq.dequeue())
print(epq.dequeue())
print(epq.dequeue())
print(epq.dequeue())
if __name__ == "__main__":
fixed_priority_queue()
element_priority_queue()

141
data_structures/queues/queue_by_list.py Normal file
View File

@ -0,0 +1,141 @@
"""Queue represented by a Python list"""
from collections.abc import Iterable
from typing import Generic, TypeVar
_T = TypeVar("_T")
class QueueByList(Generic[_T]):
def __init__(self, iterable: Iterable[_T] | None = None) -> None:
"""
>>> QueueByList()
Queue(())
>>> QueueByList([10, 20, 30])
Queue((10, 20, 30))
>>> QueueByList((i**2 for i in range(1, 4)))
Queue((1, 4, 9))
"""
self.entries: list[_T] = list(iterable or [])
def __len__(self) -> int:
"""
>>> len(QueueByList())
0
>>> from string import ascii_lowercase
>>> len(QueueByList(ascii_lowercase))
26
>>> queue = QueueByList()
>>> for i in range(1, 11):
... queue.put(i)
>>> len(queue)
10
>>> for i in range(2):
... queue.get()
1
2
>>> len(queue)
8
"""
return len(self.entries)
def __repr__(self) -> str:
"""
>>> queue = QueueByList()
>>> queue
Queue(())
>>> str(queue)
'Queue(())'
>>> queue.put(10)
>>> queue
Queue((10,))
>>> queue.put(20)
>>> queue.put(30)
>>> queue
Queue((10, 20, 30))
"""
return f"Queue({tuple(self.entries)})"
def put(self, item: _T) -> None:
"""Put `item` to the Queue
>>> queue = QueueByList()
>>> queue.put(10)
>>> queue.put(20)
>>> len(queue)
2
>>> queue
Queue((10, 20))
"""
self.entries.append(item)
def get(self) -> _T:
"""
Get `item` from the Queue
>>> queue = QueueByList((10, 20, 30))
>>> queue.get()
10
>>> queue.put(40)
>>> queue.get()
20
>>> queue.get()
30
>>> len(queue)
1
>>> queue.get()
40
>>> queue.get()
Traceback (most recent call last):
...
IndexError: Queue is empty
"""
if not self.entries:
raise IndexError("Queue is empty")
return self.entries.pop(0)
def rotate(self, rotation: int) -> None:
"""Rotate the items of the Queue `rotation` times
>>> queue = QueueByList([10, 20, 30, 40])
>>> queue
Queue((10, 20, 30, 40))
>>> queue.rotate(1)
>>> queue
Queue((20, 30, 40, 10))
>>> queue.rotate(2)
>>> queue
Queue((40, 10, 20, 30))
"""
put = self.entries.append
get = self.entries.pop
for _ in range(rotation):
put(get(0))
def get_front(self) -> _T:
"""Get the front item from the Queue
>>> queue = QueueByList((10, 20, 30))
>>> queue.get_front()
10
>>> queue
Queue((10, 20, 30))
>>> queue.get()
10
>>> queue.get_front()
20
"""
return self.entries[0]
if __name__ == "__main__":
from doctest import testmod
testmod()

115
data_structures/queues/queue_by_two_stacks.py Normal file
View File

@ -0,0 +1,115 @@
"""Queue implementation using two stacks"""
from collections.abc import Iterable
from typing import Generic, TypeVar
_T = TypeVar("_T")
class QueueByTwoStacks(Generic[_T]):
def __init__(self, iterable: Iterable[_T] | None = None) -> None:
"""
>>> QueueByTwoStacks()
Queue(())
>>> QueueByTwoStacks([10, 20, 30])
Queue((10, 20, 30))
>>> QueueByTwoStacks((i**2 for i in range(1, 4)))
Queue((1, 4, 9))
"""
self._stack1: list[_T] = list(iterable or [])
self._stack2: list[_T] = []
def __len__(self) -> int:
"""
>>> len(QueueByTwoStacks())
0
>>> from string import ascii_lowercase
>>> len(QueueByTwoStacks(ascii_lowercase))
26
>>> queue = QueueByTwoStacks()
>>> for i in range(1, 11):
... queue.put(i)
...
>>> len(queue)
10
>>> for i in range(2):
... queue.get()
1
2
>>> len(queue)
8
"""
return len(self._stack1) + len(self._stack2)
def __repr__(self) -> str:
"""
>>> queue = QueueByTwoStacks()
>>> queue
Queue(())
>>> str(queue)
'Queue(())'
>>> queue.put(10)
>>> queue
Queue((10,))
>>> queue.put(20)
>>> queue.put(30)
>>> queue
Queue((10, 20, 30))
"""
return f"Queue({tuple(self._stack2[::-1] + self._stack1)})"
def put(self, item: _T) -> None:
"""
Put `item` into the Queue
>>> queue = QueueByTwoStacks()
>>> queue.put(10)
>>> queue.put(20)
>>> len(queue)
2
>>> queue
Queue((10, 20))
"""
self._stack1.append(item)
def get(self) -> _T:
"""
Get `item` from the Queue
>>> queue = QueueByTwoStacks((10, 20, 30))
>>> queue.get()
10
>>> queue.put(40)
>>> queue.get()
20
>>> queue.get()
30
>>> len(queue)
1
>>> queue.get()
40
>>> queue.get()
Traceback (most recent call last):
...
IndexError: Queue is empty
"""
# To reduce number of attribute look-ups in `while` loop.
stack1_pop = self._stack1.pop
stack2_append = self._stack2.append
if not self._stack2:
while self._stack1:
stack2_append(stack1_pop())
if not self._stack2:
raise IndexError("Queue is empty")
return self._stack2.pop()
if __name__ == "__main__":
from doctest import testmod
testmod()

59
data_structures/queues/queue_on_pseudo_stack.py Normal file
View File

@ -0,0 +1,59 @@
"""Queue represented by a pseudo stack (represented by a list with pop and append)"""
from typing import Any
class Queue:
def __init__(self):
self.stack = []
self.length = 0
def __str__(self):
printed = "<" + str(self.stack)[1:-1] + ">"
return printed
"""Enqueues {@code item}
@param item
item to enqueue"""
def put(self, item: Any) -> None:
self.stack.append(item)
self.length = self.length + 1
"""Dequeues {@code item}
@requirement: |self.length| > 0
@return dequeued
item that was dequeued"""
def get(self) -> Any:
self.rotate(1)
dequeued = self.stack[self.length - 1]
self.stack = self.stack[:-1]
self.rotate(self.length - 1)
self.length = self.length - 1
return dequeued
"""Rotates the queue {@code rotation} times
@param rotation
number of times to rotate queue"""
def rotate(self, rotation: int) -> None:
for _ in range(rotation):
temp = self.stack[0]
self.stack = self.stack[1:]
self.put(temp)
self.length = self.length - 1
"""Reports item at the front of self
@return item at front of self.stack"""
def front(self) -> Any:
front = self.get()
self.put(front)
self.rotate(self.length - 1)
return front
"""Returns the length of this.stack"""
def size(self) -> int:
return self.length