Implementing an efficient queue in Python

Viewed 32760

I have been trying to implement a queue in Python, and I've been running into a problem.

I am attempting to use lists to implement the Queue data structure, however I can't quite figure out how to make enqueue and dequeue O(1) operations.

Every example I have seen online, seems to just append the enqueue operation and remove the first element from the list for the dequeue operation. But this would make the dequeue operation O(n) (where n is the size of the list) correct?

Is there something basic I have missed? Or do you have to use LinkedLists to implement a Queue efficiently?

import unittest

class Queue:
    def __init__(self):
        self._queue = []
        self.size = 0
        self.maxSize = 10

    def enqueue(self, item):
        if self.size < self.maxSize:
            self._queue.append(item)

    def dequeue(self):
        '''
        Removes an item from the front of the list. Remove first element of
        the array
        '''
        first = self._queue[0]
        del self._queue[0]
        return first
8 Answers

You can keep head and tail node instead of a queue list in queue class

class Node:
    def __init__(self, item = None):
        self.item = item
        self.next = None
        self.previous = None


class Queue:
    def __init__(self):
        self.length = 0
        self.head = None
        self.tail = None

    def enqueue(self, value):
        newNode = Node(value)
        if self.head is None:
            self.head = self.tail = newNode
        else:
            self.tail.next = newNode
            newNode.previous = self.tail
            self.tail = newNode
        self.length += 1

    def dequeue(self):
        item = self.head.item
        self.head = self.head.next 
        self.length -= 1
        if self.length == 0:
            self.tail = None
        return item

Queue implementation using list in Python, handling enqueue and dqueue as per inbuild queue data structure:

class queue:

    def __init__(self, max_size, size=0, front=0, rear=0):
        self.queue = [[] for i in range(5)] #creates a list [0,0,0,0,0]
        self.max_size = max_size
        self.size = size
        self.front = front
        self.rear = rear
    

    def enqueue(self, data):
        if not self.isFull():
            self.queue[self.rear] = data
            self.rear = int((self.rear + 1) % self.max_size)
            self.size += 1
        else:
            print('Queue is full')

    def dequeue(self):
        if not self.isEmpty():
            print(self.queue[self.front], 'is removed')
            self.front = int((self.front + 1) % self.max_size)
            self.size -= 1
        else:
            print('Queue is empty')

    def isEmpty(self):
        return self.size == 0

    def isFull(self):
        return self.size == self.max_size

    def show(self):
        print ('Queue contents are:')
        for i in range(self.size):
            print (self.queue[int((i+self.front)% self.max_size)])

        # driver program
        q = queue(5)
        q.enqueue(1)
        q.enqueue(2)
        q.enqueue(3)
        q.enqueue(4)
        q.enqueue(5)
        q.dequeue()
        q.show()

Here is my implementation of Queue using array, enqueue and dequeue are both O(1) operations. The implementation is based on CLRS.

class Queue:
    def __init__(self, length):
        """a queue of at most n elements using an array of n+1 element size"""
        self.length = length
        self.queue = [None]*(length+1)
        self.head = 0
        self.tail = 0

    def enqueue(self, x):
        if self.is_full():
            return 'Overflow'
        self.queue[self.tail] = x
        if self.tail == self.length:
            self.tail = 0
        else:
            self.tail = self.tail + 1

    def dequeue(self):
        if self.is_empty():
            return 'Underflow'
        x = self.queue[self.head]
        if self.head == self.length:
            self.head = 0
        else:
            self.head = self.head + 1
        return x

    def is_empty(self):
        if self.head == self.tail:
            return True
        return False

    def is_full(self):
        if self.head == self.tail+1 or (self.head == 0 and self.tail == self.length):
            return True
        return False
# Linear Queue Implementation using Array
class Queue:
    """Class with List as Array """
    def __init__(self):
        self.v_list=[]
    """Storing the in FIFO order"""
    def enqueue(self,value):
        self.v_list.append(value)
   """Removing Element from Queue in FIFO order"""
    def dequeue(self):
        if len(self.v_list)==0:
            print('Queue is Empty')
            return
        self.v_list.pop(0)

    def print_queue(self):
        print(self.v_list)

    def size_of_queue(self):
        return print(len(self.v_list))

object_queue=Queue()
object_queue.enqueue(0)
object_queue.enqueue(1)
object_queue.enqueue(2)
object_queue.enqueue(3)
object_queue.enqueue(4)
object_queue.enqueue(5)
object_queue.print_queue()
object_queue.dequeue()
object_queue.print_queue()
object_queue.dequeue()
object_queue.print_queue()
object_queue.size_of_queue()

#Circular Queue Implementation using Array
class CircularQueue():
    def __init__(self):
        """Class to hold the Postions for insert and delete"""
        self.start_pointer=0
        self.end_pointer=-1
        self.queue_list=[]
    """Storing the element in Circular order, with circular we can remove empty Block"""
    def enqueue(self,value):
        if len(self.queue_list)>10:
            print("Circular Queue is Full")
            return
        """Checking for Empty Block in Array and storing data and reseting the stat end point to process the element"""
        if 'None' in self.queue_list:
            self.queue_list[self.end_pointer]=value
            self.end_pointer+=1
        else:
            self.queue_list.append(value)
            self.end_pointer+=1
    """Removing element In FIFO order and reseting start ending point"""
    def dequeue(self):
        #self.queue_list.replace(self.queue_list[self.start_pointer],None)
        self.queue_list = [str(sub).replace(str(self.queue_list[self.start_pointer]),'None') for sub in self.queue_list] 
        self.start_pointer+=1
        for i ,j in enumerate(self.queue_list):
            if j=='None':
                self.end_pointer=i
                break
    """For Printing Queue"""            
    def print_cq(self):
        if len(self.queue_list)>10:
            print("Circular Queue is Full")
            return
        print(self.queue_list,self.start_pointer,self.end_pointer)


cir_object=CircularQueue()
cir_object.enqueue(0)
cir_object.enqueue(1)
cir_object.enqueue(2)
cir_object.enqueue(3)
cir_object.enqueue(4)
cir_object.enqueue(5)
cir_object.enqueue(6)
cir_object.enqueue(7)
cir_object.enqueue(8)
cir_object.enqueue(9)
#cir_object.print_cq()
cir_object.dequeue()
cir_object.dequeue()
cir_object.print_cq()
cir_object.enqueue(15)
cir_object.enqueue(20)
cir_object.print_cq()
class Queue:
    def __init__(self,no):
        self.no = no 
        self.SQueue = []
        self.front = -1
        self.rear = -1
    def insert(self):
        if self.rear == self.no -1:
            print("Queue is Full.....")
        else:           
            if self.front == -1:
                self.front = 0
                self.rear = 0
            else :
                self.rear += 1
            n = int(input("enter an element :: "))
            self.SQueue.insert(self.rear, n)

    def delete(self):
        if self.front == -1 and self.front == no - 1:
            print("Queue is Empty.....")
        else:
            self.SQueue.pop(self.front)
            self.front +=1
    def disp(self):
        if self.front == -1 and self.front == no - 1:
            print("Queue is Empty.....")
        else:
            print("REAR \tELEMENT")
            for i in range(len(self.SQueue)):
                print(i," \t",self.SQueue[i])

no = int(input("ENTER  Size :: "))
q = Queue(no)
while(True):
    print(" 1: INSERT ")
    print(" 2: DELETE ")
    print(" 3: PRINT ")
    print(" 4: EXIT ")
    option = int(input("enter your choice :: "))

    if option == 1:
        q.insert()
 
    elif option == 2:
        q.delete()

    elif option == 3:
        q.disp()

    elif option == 4:
        print("you are exit!!!!!")
        break
    else:
        print("Incorrect option")
 

Python's own queue implementation deque is implemented directly as C code in CPython. So it is very hard to match its performance with pure Python code. But this should not deter you from trying (unlike what @Cecil Curry said), because it makes you learn better.

Shaon shaonty implemented it as a linked list. I have implemented it using dictionaries as follows:

class Queue:
    slots = ['_dict','_max_item','_min_item','_min_reshape_countdown_length', '_reshape_countdown']
    
    def __init__(self):
        #-------- Imlementation Constants ---------
        self._min_reshape_countdown_length = 1000
        #-------------------------------------------
        
        self._dict = {}
        self._max_item = None
        self._min_item = None
        self._reshape_countdown = self._min_reshape_countdown_length
    
    def appendleft(self, item): #queue method, named to be consistent with deque
        if not self._dict: #If the queue is empty
            self._dict = {0:item}
            self._max_item = self._min_item = 0
        else:
            self._dict[self._min_item - 1] = item
            self._min_item -= 1
        
        if self._reshape_countdown == 0:
            self._reshape_countdown = max(len(self._dict), self._min_reshape_countdown_length)
            self._reshape()
        else:
            self._reshape_countdown -=1
        
    def pop(self): #dequeue method, named to be consistent with deque
        if not self._dict:
            raise IndexError('Queue is empty!')
            
        item = self._dict[self._max_item]
        del self._dict[self._max_item]
        if not self._dict: #item was the last element
            self._max_item = self._min_item = None
        else:
            self._max_item -= 1
        
        if self._reshape_countdown == 0:
            self._reshape_countdown = max(len(self._dict), self._min_reshape_countdown_length)
            self._reshape()
        else:
            self._reshape_countdown -=1
        
        return item
        
    def _reshape(self):
        if not self._dict: #if empty, no reshaping is needed
            return
        
        positives = max(0, self._max_item)
        negatives = -min(0, self._min_item)
        size = len(self._dict)
        
        #If the data is evenly distributed around 0, then reshaping is not
        #necessary.
        if positives > size/3 and negatives > size/3:
            return
        
        center = (self._min_item + self._max_item)//2
        self._dict = {key-center: value for key,value in self._dict.items()}
        self._max_item = self._max_item - center
        self._min_item = self._min_item - center
    

Note that this is not a good implementation to use for large queues, since creating large integers would be a problem. One can solve that problem by dividing the queue into blocks each having a dictionary of some fixed size.

I have compared the linked list implementation of Shaon shaonty, my implementation with dicts, and collections.deque with the following code:

import random

operations = []

queue_commands = 0
dequeue_commands = 0
for i in range(1000000):
    if queue_commands > dequeue_commands:
        operations.append(('q',random.random()) if random.random() < 1/2 else ('d',None))
    else:
        operations.append(('q',random.random()))
    
    if operations[-1][0] == 'q':
        queue_commands += 1
    else:
        dequeue_commands += 1

def test_case(queue):
    for command, element in operations:
        if command == 'q':
            queue.appendleft(element)
        else:
            queue.pop()
    

the results are

%timeit test_case(Linked_Queue())
796 ms ± 2.13 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

%timeit test_case(Queue())
838 ms ± 1.5 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

%timeit test_case(deque())
120 ms ± 566 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)

As I said, it is not easy to beat deque with raw Python code:

class Dummy:

    def appendleft(self,item):
        pass

    def pop(self):
        pass

%timeit test_case(Dummy())
176 ms ± 327 µs per loop (mean ± std. dev. of 7 runs, 10 loops each)

In dequeue method there is no any loop. You only have to do list operations. Therefore the time complexity for dequeue is also O(n)(linear).

class Queue:
    def __init__(self):
       self.items=[]
    def enqueue(self,item):
       self.items.append(item)
    def dequeue(self):
       return self.items.pop(0)
    def isEmpty(self):
       return self.items==[]
    def __len__(self):
       return len(self.items)
Related