Creating a priority queue using a heap in Python

Question

I am new to Python so excuse for the silly mistakes... I am trying to create a priority queue using heap in python(2.7.15) and my code doesn't obviously work.

pq = []                         # list of entries arranged in a heap
entry_finder = {}               # mapping of tasks to entries
REMOVED = '<removed-task>'      # placeholder for a removed task
count = 0     # unique sequence count

def push(pq,task,priority=0):
    'Add a new task'
    count = count+1
    entry = [priority, count, task]
    entry_finder[task] = entry
    heappush(pq, entry)

def update(pq,task, priority=0):
    'Add a new task or update the priority of an existing task'
    if task in entry_finder:
        remove_task(task)
    count = count+1
    entry = [priority, count, task]
    entry_finder[task] = entry
    heappush(pq, entry)

def remove_task(task):
    'Mark an existing task as REMOVED.  Raise KeyError if not found.'
    entry = entry_finder.pop(task)
    entry[-1] = REMOVED

def pop(pq):
    'Remove and return the lowest priority task. Raise KeyError if empty.'
    while pq:
        priority, count, task = heappop(pq)
        if task is not REMOVED:
            del entry_finder[task]
            return task
    raise KeyError('pop from an empty priority queue')

def IsEmpty(pq):
    if not pq:
    print("List is empty")

Thats what I have done,most of them are taken by here: https://docs.python.org/2/library/heapq.html. My problem is when I try to run it on the python interprenter,I get this:

>>> pq=[]
>>> pq.push("task1",1)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: 'list' object has no attribute 'push'

My question is what can I do to avoid this error and if my code has any flaws that maybe can cause further errors?

Answer 1

You're building a class in an "object oriented C"-style that involves passing the object context into the function, e.g. do_something(instance, arg). This is possible, but isn't natural to Python, which supports classes and a keyword called self that represents an instance of an object, allowing one to write instance.do_something(arg) as you did when you called your push function.

With this approach, a class would encapsulate all of the state data you currently have in the global scope:

pq = []                         # list of entries arranged in a heap
entry_finder = {}               # mapping of tasks to entries
REMOVED = '<removed-task>'      # placeholder for a removed task
count = 0                       # unique sequence count

Even as written in "C"-style, the program has no instance state outside of these global variables; you'd need some kind of structure to keep these variables together and pass them into functions, or use the global keyword for each variable inside each of your functions, which is not a very good solution in terms of state safety, reusability, understandability or any other design metric.

Here's an example of the program refactored as a class:

from heapq import heappush, heappop


class PQ:
    def __init__(self):
        self.pq = []
        self.entry_finder = {}
        self.REMOVED = '<removed-task>'
        self.count = 0

    def push(self, task, priority=0):
        '''Add a new task
        '''
        self.count += 1
        entry = [priority, self.count, task]
        self.entry_finder[task] = entry
        heappush(self.pq, entry)

    def update(self, task, priority=0):
        '''Add a new task or update the priority of an existing task
        '''
        if task in self.entry_finder:
            self.remove_task(task)

        self.count += 1
        entry = [priority, self.count, task]
        self.entry_finder[task] = entry
        heappush(self.pq, entry)

    def remove_task(self, task):
        '''Mark an existing task as REMOVED.  Raise KeyError if not found.
        '''
        entry = self.entry_finder.pop(task)
        entry[-1] = self.REMOVED

    def pop(self):
        '''Remove and return the lowest priority task. Raise KeyError if empty.
        '''
        while self.pq:
            priority, count, task = heappop(self.pq)

            if task is not self.REMOVED:
                del self.entry_finder[task]
                return task

        raise KeyError('pop from an empty priority queue')

    def empty(self):
        return len(self.pq) == 0

Having done this, you can now import the class in your interpreter (make sure the source code is in the same folder, pq.py), create an instance of the class and begin using it:

>>> from pq import PQ
>>> pq = PQ()
>>> pq.push(1, 20)
>>> pq.push(2, 30)
>>> pq.push(3, 10)
>>> while not pq.empty(): print pq.pop()
...
3
1
2
>>>

Another common practice is to add testing right into the class file using the if __name__ == '__main__': conditional:

if __name__ == '__main__':
    pq = PQ()
    pq.push(1, 20)
    pq.push(2, 30)
    pq.push(3, 10)

    while not pq.empty():
        print pq.pop()

This can be run on the terminal with python pq.py.

Try it!

Answer 2

The way you have defined the push function, you need to pass the list representing the queue as the first argument:

Instead of

pq = []
pq.push("task1", 1)

do

pq = []
push(pq, "task1", 1)