Skip to content

Implementing a Queue in Python: A Step-by-Step Guide

Updated: at 03:34 AM

A queue is a fundamental data structure used in many areas of computer science and programming. It follows the first-in, first-out (FIFO) principle, meaning items are added to the end of the queue and removed from the front. Queues provide efficient order processing and are commonly implemented using arrays or linked lists.

In this comprehensive guide, we will walk through the process of building a queue from scratch in Python using lists and discuss key concepts related to queue operations and applications.

Table of Contents

Open Table of Contents

Overview of Queues

A queue represents a linear data structure that models real-world queueing systems, like a line of customers waiting to be served. The essential operations of a queue are:

Queues maintain a FIFO ordering, meaning the first element added will be the first one removed. This differs from stacks, which follow last-in, first-out (LIFO).

Queues are useful when the order of operations matters, like:

In Python, queues can be implemented using lists or the collections.deque object. We will focus on building one from scratch with lists.

Implementing a Queue in Python

We will create a Queue class containing the key methods enqueue and dequeue.

class Queue:
  def __init__(self):
    self.items = []

  def enqueue(self, item):
    self.items.append(item)

  def dequeue(self):
    if self.isEmpty():
      return "Queue is empty"
    return self.items.pop(0)

  def peek(self):
    if self.isEmpty():
      return "Queue is empty"
    return self.items[0]

  def isEmpty(self):
    return self.items == []

  def size(self):
    return len(self.items)

This provides a basic queue implementation we can start using:

q = Queue()
q.enqueue(1)
q.enqueue(2)
q.enqueue(3)

print(q.size()) # 3

print(q.dequeue()) # 1
print(q.dequeue()) # 2

print(q.peek()) # 3

However, this queue implementation has some limitations:

Next, we will go over approaches to address these.

Queue Operations and Time Complexity

The time complexity of queue operations using a list is:

The pop(0) method for dequeuing leads to poor performance as the size grows since all elements must be shifted down after removing the front item.

We can improve the performance by using a collections.deque object which was designed to efficiently add and remove from both ends:

from collections import deque

queue = deque()

The time complexity then becomes:

However, for implementing from scratch, we can make our list-based queue more efficient by reversing the list when an item is dequeued:

def dequeue(self):
  if self.isEmpty():
    return "Queue is empty"

  front = self.items.pop()

  if len(self.items) > 0:
    self.items.reverse()

  return front

By reversing after each dequeue, the old front element gets moved to the end. Reversing a list is O(n) but only has to be done occasionally.

Adding a Maximum Capacity

For many real-world queues, there is a limit to how many items can be enqueued before the queue is considered full.

We can add a maxsize parameter to check capacity when enqueueing:

class Queue:
  def __init__(self, maxsize=10):
    self.items = []
    self.maxsize = maxsize

# Rest of methods...

  def enqueue(self, item):
    if self.size() == self.maxsize:
      return "Queue is full!"
    self.items.append(item)

Now if the queue size reaches the max, enqueueing any more items will return an error.

Making the Queue Thread Safe

Since our queue is using a shared list between threads, we need to add thread safety to avoid race conditions between enqueueing and dequeueing.

We can use locks from the threading module:

import threading

class Queue:
  def __init__(self):
    self.items = []
    self.lock = threading.Lock()

  def enqueue(self, item):
    with self.lock:
      self.items.append(item)

  def dequeue(self):
    with self.lock:
      if self.isEmpty():
        return "Queue is empty"
      return self.items.pop(0)

By wrapping critical sections in with self.lock, it forces threads to wait their turn before accessing the shared resource.

This prevents data corruption from race conditions when multiple threads are competing to edit the queue.

Applications of Queues

Now that we have a basic thread safe queue implementation, let’s look at some examples of using it:

1. Priority Job Queue

class Job:
  def __init__(self, priority, description):
    self.priority = priority
    self.description = description

job_queue = Queue()

job1 = Job(1, "Process data")
job2 = Job(3, "Print documents")
job3 = Job(2, "Backup files")

# Enqueue jobs by priority
job_queue.enqueue(job1)
job_queue.enqueue(job3)
job_queue.enqueue(job2)

# Print jobs in queue order
while not job_queue.isEmpty():
  next_job = job_queue.dequeue()
  print(next_job.description)

# Output:
# Process data
# Backup files
# Print documents

Here we create a Job class that holds a priority and description. Jobs are enqueued based on priority and processed in a first-in, first-out order.

2. Circular Queue for Fixed-Size Buffer

# Producer thread
def producer(buffer, item):
  buffer.enqueue(item)

# Consumer thread
def consumer(buffer):
  while True:
    item = buffer.dequeue()
    if item is not None:
      # Process item
      print(item)
    else:
      # Sleep to allow more items to be enqueued
      time.sleep(0.1)

# Create buffer
buffer = Queue(maxsize=5)

# Start and run threads continuously
producer_thread = threading.Thread(target=producer, args=(buffer, 1))
consumer_thread = threading.Thread(target=consumer, args=(buffer,))

producer_thread.start()
consumer_thread.start()

Here a circular queue is used to share items between a producer and consumer thread safely with fixed buffer size. The threads run continuously to process an infinite stream.

Summary

We have now built a fully-featured queue in Python complete with enqueue, dequeue, peeking, capacity limiting, and thread safety.

Key takeaways include:

Queues are a versatile data structure for a wide range of algorithms and programs. This guide covered common techniques for optimization and thread safety when implementing queues in Python.