Queues implementation using Python

July 31, 2018 4 min read Python Queues

Queue A queue is a collection of objects in which elements are added/deleted based on the principle of FIFO (First In First Out)

Queue ADT :-

Q.enqueue(e) : Add an element to the end

Q.dequeue() : Remove and return the first element, an error will be reported if empty

Q._front() : Returns the first element without removing it, an error will be reported if empty

Q.isEmpty() : Returns True if the Queue is empty

Q.isFull() : Returns True if the Queue is full

Q.size() : Returns the length of Queue

Implementation

Queue will be implemented using Lists (circular arrays), where we assume that the Queue is of fixed size N

Test Code

class FullQueueException(Exception):
    pass

class EmptyQueueException(Exception):
    pass

class Queue(object):
    def __init__(self,N=10):
        self.N = N  # Default size of queue = 10
        self._A = [None] * self.N
        self._front = -1
        self._rear = -1

    def isEmpty(self):
        if (self._front == -1) and (self._rear == -1):
            return True
        else:
            return False

    def isFull(self):
        return ((self._rear + 1) % self.N) == self._front

    def enqueue(self,e):
        if self.isEmpty():
            self._front = 0
            self._rear = 0
        elif self.isFull():
            raise FullQueueException('Queue is full')
        else:
            self._rear = (self._rear+1) % self.N
        self._A[self._rear] = e

    def dequeue(self):
        if self.isEmpty():
            raise EmptyQueueException('Queue is empty')
        elif (self._front == self._rear):
            self._A[self._front] = None # Don't care about the data but this is for garbage collector 
            self._front = -1
            self._rear = -1
        else:
            self._A[self._front] = None # Don't care about the data but this is for garbage collector
            self._front = (self._front+1) % self.N
    
    def size(self):
        if self.isEmpty():
            return 0
        return ((self.N - self._front + self._rear) % self.N) + 1

    def front(self):
        if self.isEmpty():
            raise EmptyQueueException('Queue is empty')
        return self._A[self._front]
import unittest
from queues import Queue, FullQueueException, EmptyQueueException

class TestQueues(unittest.TestCase):
    def setUp(self):
        self.Q = Queue(N=5)
  
    def test_setup(self):
        "Test for Empty Queue"
        self.assertEqual(self.Q.size(),0)
        self.assertTrue(self.Q.isEmpty())
        self.assertFalse(self.Q.isFull())
        with self.assertRaises(EmptyQueueException) as cm:
            self.Q.dequeue()
        expected_msg = "Queue is empty"
        self.assertEquals(cm.exception.message, expected_msg)
        with self.assertRaises(EmptyQueueException) as cm:
            self.Q.front()
        expected_msg = "Queue is empty"
        self.assertEquals(cm.exception.message, expected_msg)

    def test_enqueue_dequeue(self):
        self.Q.enqueue('A')
        self.Q.enqueue('B')
        self.assertEqual(self.Q.size(),2)
        self.assertFalse(self.Q.isEmpty())
        self.assertFalse(self.Q.isFull())
        self.Q.enqueue('C')
        self.Q.enqueue('D')
        self.Q.enqueue('E')
        self.assertFalse(self.Q.isEmpty())
        self.assertTrue(self.Q.isFull())
        self.assertEqual(self.Q.front(),'A')
        with self.assertRaises(FullQueueException) as cm:
            self.Q.enqueue('F')
        expected_msg = "Queue is full"
        self.assertEquals(cm.exception.message, expected_msg)
        self.assertEqual(self.Q.size(),5)
        self.Q.dequeue()
        self.assertEqual(self.Q.size(),4)
        self.assertFalse(self.Q.isEmpty())
        self.assertFalse(self.Q.isFull())
        self.Q.enqueue('G')
        self.assertEqual(self.Q.front(),'B')
        self.assertEqual(self.Q.size(),5)
        self.Q.dequeue()
        self.Q.dequeue()
        self.Q.dequeue()
        self.Q.dequeue()
        self.assertEqual(self.Q.size(),1)
        self.assertEqual(self.Q.front(),'G')
        self.assertFalse(self.Q.isEmpty())
        self.assertFalse(self.Q.isFull())
        self.Q.dequeue()
        self.assertEqual(self.Q.size(),0)
        with self.assertRaises(EmptyQueueException) as cm:
            self.Q.dequeue()
        expected_msg = "Queue is empty"
        self.assertEquals(cm.exception.message, expected_msg)
        self.assertTrue(self.Q.isEmpty())
        self.assertFalse(self.Q.isFull())

if __name__ == '__main__':
    unittest.main(verbosity=2)
class FullQueueException(Exception):
    pass

class EmptyQueueException(Exception):
    pass

class Queue(object):
    def __init__(self,N=10):
        self.N = N  # Default size of queue = 10
        self._A = [None] * self.N
        self._front = -1
        self._rear = -1

    def isEmpty(self):
        if (self._front == -1) and (self._rear == -1):
            return True
        else:
            return False

    def isFull(self):
        return ((self._rear + 1) % self.N) == self._front

    def enqueue(self,e):
        if self.isEmpty():
            self._front = 0
            self._rear = 0
        elif self.isFull():
            raise FullQueueException('Queue is full')
        else:
            self._rear = (self._rear+1) % self.N
        self._A[self._rear] = e

    def dequeue(self):
        if self.isEmpty():
            raise EmptyQueueException('Queue is empty')
        elif (self._front == self._rear):
            self._A[self._front] = None # Don't care about the data but this is for garbage collector 
            self._front = -1
            self._rear = -1
        else:
            self._A[self._front] = None # Don't care about the data but this is for garbage collector
            self._front = (self._front+1) % self.N
    
    def size(self):
        if self.isEmpty():
            return 0
        return ((self.N - self._front + self._rear) % self.N) + 1

    def front(self):
        if self.isEmpty():
            raise EmptyQueueException('Queue is empty')
        return self._A[self._front]
import unittest
from queues import Queue, FullQueueException, EmptyQueueException

class TestQueues(unittest.TestCase):
    def setUp(self):
        self.Q = Queue(N=5)
  
    def test_setup(self):
        "Test for Empty Queue"
        self.assertEqual(self.Q.size(),0)
        self.assertTrue(self.Q.isEmpty())
        self.assertFalse(self.Q.isFull())
        with self.assertRaises(EmptyQueueException) as cm:
            self.Q.dequeue()
        expected_msg = "Queue is empty"
        self.assertEquals(cm.exception.message, expected_msg)
        with self.assertRaises(EmptyQueueException) as cm:
            self.Q.front()
        expected_msg = "Queue is empty"
        self.assertEquals(cm.exception.message, expected_msg)

    def test_enqueue_dequeue(self):
        self.Q.enqueue('A')
        self.Q.enqueue('B')
        self.assertEqual(self.Q.size(),2)
        self.assertFalse(self.Q.isEmpty())
        self.assertFalse(self.Q.isFull())
        self.Q.enqueue('C')
        self.Q.enqueue('D')
        self.Q.enqueue('E')
        self.assertFalse(self.Q.isEmpty())
        self.assertTrue(self.Q.isFull())
        self.assertEqual(self.Q.front(),'A')
        with self.assertRaises(FullQueueException) as cm:
            self.Q.enqueue('F')
        expected_msg = "Queue is full"
        self.assertEquals(cm.exception.message, expected_msg)
        self.assertEqual(self.Q.size(),5)
        self.Q.dequeue()
        self.assertEqual(self.Q.size(),4)
        self.assertFalse(self.Q.isEmpty())
        self.assertFalse(self.Q.isFull())
        self.Q.enqueue('G')
        self.assertEqual(self.Q.front(),'B')
        self.assertEqual(self.Q.size(),5)
        self.Q.dequeue()
        self.Q.dequeue()
        self.Q.dequeue()
        self.Q.dequeue()
        self.assertEqual(self.Q.size(),1)
        self.assertEqual(self.Q.front(),'G')
        self.assertFalse(self.Q.isEmpty())
        self.assertFalse(self.Q.isFull())
        self.Q.dequeue()
        self.assertEqual(self.Q.size(),0)
        with self.assertRaises(EmptyQueueException) as cm:
            self.Q.dequeue()
        expected_msg = "Queue is empty"
        self.assertEquals(cm.exception.message, expected_msg)
        self.assertTrue(self.Q.isEmpty())
        self.assertFalse(self.Q.isFull())

if __name__ == '__main__':
    unittest.main(verbosity=2)

Related Posts

Selenium ActionChains using Python with examples

ActionChains are used to automate interactions like mouse movement, mouse button action, key press, context menu, drag and drop etc. These are typical...

June 21, 2021 Read →

Quick tutorial with Selenium using Python

This post will be quick beginner's tutorial for Selenium using Python binding. First step is to install "python-selenium" and get the required browser...

March 27, 2021 Read →

Comments