Queue: Python Implementation

What is the Queue?

The queue is an abstract linear data structure.

The queue is a first in first out data structure, which means that item that the item that goes in first is the item that comes out first. The queue natively does not support index-based access, unlike the array. The new element can only be added at the back of the queue and can be removed only from the front of the queue. The queue is a widely used data structure in various day-to-day applications. The queue is used in the implementation of operating system kernel implementation and also in the language compiler implementation.

Pros of the Queue

  1. Constant time insertion and deletion of elements in FIFO order
  2. Dynamic memory allocation, unlike array. Assuming linked list if used for implementation.

Cons of the Queue

  1. Only single-time traversal is possible
  2. Doesn’t support searching

Applications of the Queue

  1. Scheduling of processes by the operating system, scheduling of jobs in distributed systems, scheduling of requests in a web server, etc.
  2. Breath first searches in graph and tree search.
  3. Handling hardware and software interrupts.

Queue Operations

Some common operations in the queue are:

  1. Enqueue/push: Add a new element at the back of the queue.
    a. For efficiency enqueue operation should be of constant complexity.
  2. Dequeue/pop: Remove an element from the front of the queue and returns it.
    a. For efficiency dequeue operation should be of constant complexity.
  3. Peek: Returns the front element without removing it.
  4. is_empty: Check if the queue is empty
  5. Is_full: Check if the queue is full.

Queue Implementations

Python natively provides Queue implementation and can be accessed by importing the queue module or using the deque module from collections.

The queue can also be implemented using a linked list. All the operation can be implemented in O(1) complexity, only overhead is the memory to store the next pointer of each node. Total extra memory requirement is O(n). Linked list is suitable when you need fast insertion and deletion and memory usage is not an issue.

Here we will show how to implement the queue using a linked list.

Let’s define a node for the linked list:

from typing import Any

class Node:
  """Node of a linked list.

  Args:
    value: Value of the node.
  """
  def __init__(self, value: Any):
    self.value = value
    self.next = None

Using this node as an element of the queue we will implement the basic operations.

Enqueue or push operation

Enqueue operation adds an element at the back of the queue. In a linked list this can be implemented in constant complexity using an additional pointer that holds the address of the last node. We define a variable to hold the last node address in the Queue constructor. Apart from these here we will also define the maximum size of the queue, there is no overhead of this parameter as in this implementation we don’t need to allocate fixed memory as the linked list is dynamic. When using an array to implement the queue, we will need to allocate fixed memory, though a dynamic array has its advantages in terms of random access.

class Queue:
  """Queue implementation using linked list.

  Args:
    max_size: Max size of the queue.
  """

  def __init__(self, max_size: int):
    """Set queue instance attributes."""
    self.max_size = max_size
    self.size = 0
    # first node or the head of the list
    self.start = None
    # last node of the list
    self.end = None 

Adds an element in the queue, here this operation is of constant complexity O(1).

  def enqueue(self, value: Any) -> None:
    """Add a new item to the queue.

    Args:
      value: Value to add.

    Raises:
      IOError: If the queue has reached the maximum allowed size.
    """
    if self.size >= self.max_size:
      raise IOError('Size overflowed')
    node = Node(value)
    if self.end:
      self.end.next = node
      self.end = node
    else:
      self.end = node
      self.start = node
    self.size += 1

Dequeue or pop operation

Removes an element from the front of the queue, here this operation is of constant complexity O(1).

  def dequeue(self) -> Any:
    """Remove an item from the queue.

    Returns:
      The removed value if the queue is not empty else None.
    """
    if not self.start:
      return None
    value = self.start.value
    self.start = self.start.next
    self.size -= 1
    return value

Peek operation

Get the element at the front of the queue, this is also of constant complexity O(1).

  def peek(self) -> Any:
    """Get the item at the front of the queue.

    Returns:
      Value at the front of the queue.
    """
    if self.start:
      return self.start.value
    return -1

is_empty operation

Check if the queue is empty, this is also of constant complexity O(1).

  def is_empty(self) -> bool:
    """Check if the queue is empty.

    Returns:
      A bool indicates whether the queue is empty or not.
    """
    if self.size < 1:
      return True
    return False

is_full operation

Check if the queue is full, this is also of constant complexity O(1).

  def is_full(self) -> bool:
    """Check if the queue is full.

    Returns:
      A bool indicating whether the queue is full or not.
    """
    if self.size == self.max_size:
      return True
    return False

Combining everything:

from typing import Any

class Node:
  """Node of a linked list.

  Args:
    value: Value of the node.
  """
  def __init__(self, value: Any):
    self.value = value
    self.next = None


class Queue:
  """Queue implementation using linked list.

  Args:
    max_size: Max size of the queue.
  """

  def __init__(self, max_size: int):
    """Set queue instance attributes."""
    self.max_size = max_size
    self.size = 0
    self.start = None
    self.end = None

  def enqueue(self, value: Any) -> None:
    """Add a new item to the queue.

    Args:
      value: Value to add.

    Raises:
      IOError: If the queue has reached the maximum allowed size.
    """
    if self.size >= self.max_size:
      raise IOError('Size overflowed')
    node = Node(value)
    if self.end:
      self.end.next = node
      self.end = node
    else:
      self.end = node
      self.start = node
    self.size += 1

  def dequeue(self) -> Any:
    """Remove an item from the queue.

    Returns:
      The removed value if the queue is not empty else None.
    """
    if not self.start:
      return None
    value = self.start.value
    self.start = self.start.next
    self.size -= 1
    return value

  def peek(self) -> Any:
    """Get the item at the front of the queue.

    Returns:
      Value at the front of the queue.
    """
    if self.start:
      return self.start.value
    return -1

  def is_empty(self) -> bool:
    """Check if the queue is empty.

    Returns:
      A bool indicates whether the queue is empty or not.
    """
    if self.size < 1:
      return True
    return False

  def is_full(self) -> bool:
    """Check if the queue is full.

    Returns:
      A bool indicating whether the queue is full or not.
    """
    if self.size == self.max_size:
      return True
    return False


if __name__ == '__main__':
  q = Queue(3)

  q.enqueue(1)
  print(q.end.value)
  q.enqueue(2)
  print(q.end.value)
  q.enqueue(3)
  print(q.end.value)

  print(q.is_full())
  print(q.is_empty())
  print(q.dequeue())
  print(q.dequeue())
  print(q.dequeue())

Implementation of the queue using python deque modoule

Python provides a double ended queue implementation called deque. Here we can perform insertion/deletion from both end of the queue.
Python deque can be used to implement queue, here also all operations are of O(1) complexity.

deque module popleft method is used to remove element from the front of the queue in O(1) complexity, and append method is used to add element at the back of the queue in O(1).

Random access of the queue elements are possible in this implementation, though the complexity of accessing the elements in the middle is O(n). For more details visit: collections — Container datatypes — Python 3.12.0 documentation

Here is the full implementation

from collections import deque
from typing import Any

class Queue:
  """Queue implementation using linked list.

  Args:
    max_size: Max size of the queue.
  """

  def __init__(self, max_size: int):
    """Set queue instance attributes."""
    self.max_size = max_size
    self.size = 0
    self._data = deque()

  def enqueue(self, item):
    """Add a new item to the queue.

    Args:
      value: Value to add.

    Raises:
      IOError: If the queue has reached the maximum allowed size.
    """
    if self.size >= self.max_size:
      raise IOError('Size overflowed')
    self._data.append(item)
    self.size += 1

  def dequeue(self) -> Any:
    """Remove a item form the queue.

    Returns:
      The removed value if the queue is not empty else None.
    """
    if self.size < 1:
      return None
    self.size -= 1
    return self._data.popleft()

  def peek(self) -> Any:
    """Get the item at the front of the queue.

    Returns:
      Value at the front of the queue.
    """
    if self._data:
      return self._data[0]
    return -1

  def is_empty(self) -> bool:
    """Check if the queue is empty.

    Returns:
      A bool indicates whether the queue is empty or not.
    """
    if self.size < 1:
      return True
    return False

  def is_full(self) -> bool:
    """Check if the queue is full.

    Returns:
      A bool indicating whether the queue is full or not.
    """
    if self.size == self.max_size:
      return True
    return False

  def get(self, index: int) -> Any:
    """Return the value at an index.

    Args:
      index: Index of the value to return.

    Returns:
      Value at index f exist.

    Raises:
      Exception: If vale does not exists.
    """
    try:
      return self._data[index]
    except IndexError as exc:
      raise Exception("Value does not exist at index: {index}.") from exc

if __name__ == '__main__':
  q = Queue(3)

  q.enqueue(1)
  q.enqueue(2)
  q.enqueue(3)
  print(q.peek())
  print(q.get(2))
  print(q.is_full())
  print(q.is_empty())
  print(q.dequeue())
  print(q.dequeue())
  print(q.dequeue())

Implementation of Queue using Circular Queue

A circular queue is an implementation of a queue with a maximum size that will continue to loop back over itself in a circular motion. When popping elements are not actually removed but the entry is termed as free for use when adding new elements.

A circular queue is also termed a ring buffer which is used for scheduling, memory management, data stream buffering, etc.

Random access at constant time O(1) is possible in a circular queue.

One limitation is that it has a fixed size, increasing the size is not supported in its original form, though it can be implemented with additional complexity.

To implement a circular queue we can use an array. To get the index to add or remove an element from the queue we can use the modulo operator. Using the module operator it is possible to loop the array in a circular motion.

Suppose we are at index 9 (index of the most recent element in the queue) and the capacity / maximum size of the queue is 10. Index 0, and 1 are free, to add a new element in the queue, as per the circular queue design we can add it at index 0. To get the index we can modulo operator

Index = (recent_index + 1) % max_size
= (9 + 1) % 10 == 0

The same approach is used to perform the pop operation.
The following Gif depicts the working of circular queue. The back pointer is updated when new element is added to the queue, and the front pointer is updated with an element is removed from the queue.
Both the pointer is updated using the following equation:

front = (front + 1) % self.max_size
back = (back + 1) % self.max_size

ezgif.com-gif-maker

Below is the implementation of the Queue using Circular Queue.

from typing import Any

class Queue:
  """Queue implementation using linked list.

  Args:
    max_size: Max size of the queue.
  """

  def __init__(self, max_size: int):
    """Set queue instance attributes."""
    self.max_size = max_size
    self.size = 0
    # We could use array here, but it has a limitation of homogeneous data type requirement. 
    # self._data = array.array("i", [0] * max_size)
    self._data = [0] * max_size
    self.back = -1
    self.front = 0

  def enqueue(self, item):
    """Add a new item to the queue.

    Args:
      value: Value to add.

    Raises:
      IOError: If the queue has reached the maximum allowed size.
    """
    if self.size >= self.max_size:
      raise IOError('Size overflowed')
    self.back = (self.back + 1) % self.max_size
    self._data[self.back] = item
    self.size += 1

  def dequeue(self) -> Any:
    """Remove a item form the queue.

    Returns:
      The removed value if the queue is not empty else None.
    """
    if self.size < 1:
      return None
    self.size -= 1
    popped_value = self._data[self.front]
    self.front = (self.front + 1) % self.max_size
    return popped_value

  def peek(self) -> Any:
    """Get the item at the front of the queue.

    Returns:
      Value at the front of the queue.
    """
    if self._data:
      return self._data[self.front]
    return -1

  def is_empty(self) -> bool:
    """Check if the queue is empty.

    Returns:
      A bool indicates whether the queue is empty or not.
    """
    if self.size < 1:
      return True
    return False

  def is_full(self) -> bool:
    """Check if the queue is full.

    Returns:
      A bool indicating whether the queue is full or not.
    """
    if self.size == self.max_size:
      return True
    return False

  def get(self, index: int) -> Any:
    """Return the value at an index.

    Args:
      index: Index of the value to return.

    Returns:
      Value at index if exist.

    Raises:
      Exception: If vale does not exists.
    """
    try:
      return self._data[index]
    except IndexError as exc:
      raise Exception("Value does not exist at index: {index}.") from exc


if __name__ == '__main__':
  q = Queue(3)

  q.enqueue(1)
  q.enqueue(2)
  q.enqueue(3)
  print(q.peek())

  print(q.get(2))
  print(q.get(1))

  print(q.is_full())
  print(q.is_empty())
  print(q.dequeue())
  print(q.dequeue())
  print(q.dequeue())