2
\$\begingroup\$

I've implemented an updateable priority queue as a part of implementing some single-source-shortest-path algorithms (Dijkstra, BF). I'm relatively new to Python.

class UpdateableQueue:
    def __init__(self,iterable=None):
        self._heap = []
        self._entry_finder = {}
        if iterable:
            for item in iterable:
                self._entry_finder[item[0]] = item[1]
                heapq.heappush(self._heap,(item[1],item[0]))

    def __getitem__(self, key):
        if key in self._entry_finder:
            return self._entry_finder[key]
        else:
            raise KeyError('Item not found in the priority queue')

    def __len__(self):
        return len(self._entry_finder)

    def __contains__(self, key):
        return key in self._entry_finder

    def push(self, key,priority):
        self._entry_finder[key] = priority
        heapq.heappush(self._heap, (priority, key))

    def pop(self):
        if not self._heap:
            raise IndexError("The heap is empty")

        value,key = self._heap[0]
        while key not in self or self._entry_finder[key] != value:
            heapq.heappop(self._heap)
            if not self._heap:
                raise IndexError("The heap is empty")
            value,key = self._heap[0]

        value, key = heapq.heappop(self._heap)
        del self._entry_finder[key]
        return key,value

Edit: a simple test case example of the UpdateableQueue

from UpdateableQueue.UpdateableQueue import UpdateableQueue

if __name__ == "__main__":
    # Initalizing empty queue
    queue = UpdateableQueue()
    # Inserting key=1,priority=4 element to the queue the heap is now of size 1, and the dict size 1
    queue.push(1,4)
    # editing key=1,priority=1 element to the queue the heap is now of size 2, and the dict size 1
    queue.push(1,1)
    # editing key=1,priority=2 element to the queue the heap is now of size 3, and the dict size 1
    queue.push(1,2)
    assert len(queue) == 1 # True
    assert 1 in queue # True
    # on pop we remove (key=1,value=1) from the heap and then pop out that value of (1,2), followed by
    # removing that key from the dict
    key, value = queue.pop()
    assert 1 not in queue #True
    assert key == 1 # True
    assert value == 2 # True
    # Inserting key=2,priority=5 element to the queue the heap is now of size 2, and the dict size 1
    queue.push(2,5)
    assert len(queue) == 1
    # We "clean" the remaining (1,4) element from the heap and return pop (2,5) as expected.
    key, value = queue.pop()
    assert key == 2  # True
    assert value == 5  # True
\$\endgroup\$
  • 2
    \$\begingroup\$ Could you also show this priority queue in action? \$\endgroup\$ – dfhwze Jul 27 at 13:36
  • 3
    \$\begingroup\$ @dfhwze Ive added a simple code example to it, let me know if there's anything else I can do. \$\endgroup\$ – Omri Shneor Jul 27 at 14:17
2
\$\begingroup\$

Your code looks good, there's not much for me to review, but here are a couple improvements/changes I would make.

Improvements

  • Docstrings: You should include a docstring at the beginning of every method/class/module you write. This will help any documentation catch what your code is supposed to accomplish.
  • If/Else: If you return a value in the if, then you don't need an else. You should just put the else code after the if, instead of inside an else, like so:
if key in self._entry_finder:
    return self._entry_finder[key]
else:
    raise KeyError('Item not found in the priority queue')

to

if key in self._entry_finder:
    return self._entry_finder[key]
raise KeyError('Item not found in the priority queue')

Updated Code

class UpdateableQueue:
    """ An updateable priority queue class """
    def __init__(self, iterable=None):
        self._heap = []
        self._entry_finder = {}
        if iterable:
            for item in iterable:
                self._entry_finder[item[0]] = item[1]
                heapq.heappush(self._heap, (item[1], item[0]))

    def __getitem__(self, key):
        """
        Returns the item with the specified key, if exists. Else,
        it raises a `KeyError` exception
        """
        if key in self._entry_finder:
            return self._entry_finder[key]
        raise KeyError('Item not found in the priority queue')

    def __len__(self):
        """ Returns the length of the queue """
        return len(self._entry_finder)

    def __contains__(self, key):
        """ Returns a boolean based on if the key is in the queue """
        return key in self._entry_finder

    def push(self, key, priority):
        """ Pushses a priority into the queue """
        self._entry_finder[key] = priority
        heapq.heappush(self._heap, (priority, key))

    def pop(self):
        """ Removes a priority from the queue """
        if not self._heap:
            raise IndexError("The heap is empty")

        value, key = self._heap[0]
        while key not in self or self._entry_finder[key] != value:
            heapq.heappop(self._heap)
            if not self._heap:
                raise IndexError("The heap is empty")
            value, key = self._heap[0]

        value, key = heapq.heappop(self._heap)
        del self._entry_finder[key]
        return key, value
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.