Queue

from queue import Queue

q = Queue()
q.put('a')
q.put('b')
q.put('c')
print(q.get())
print(q.get())
print(q.get())

LIFO Queue

from queue import LifoQueue

q = LifoQueue()
q.put('a')
q.put('b')
q.put('c')
print(q.get())
print(q.get())
print(q.get())

Priority queue

from queue import PriorityQueue

# The default behavior is a min_queue
min_queue = PriorityQueue()
min_queue.put((5, 'f'))
min_queue.put((2, 'h'))
min_queue.put((3, 'a'))
min_queue.put((4, 'c'))
print(min_queue.get(), 'is (2, h)')
print(min_queue.get(), 'is (3, a)')
print(min_queue.get(), 'is (4, c)')
print(min_queue.get(), 'is (5, f)')

For max-heap

PriorityQueue uses a Min heap property. Flip the sign on the priority to get a Max heap property:

from queue import PriorityQueue
from dataclasses import dataclass, field
from typing import Any

min_queue = PriorityQueue()
min_queue.put((1, ('a', 'b')))
min_queue.put((-10, ('c', 'd')))
min_queue.put((3, ('e', 'f')))
min_queue.put((2, ('g', 'h')))
print('min_queue:', min_queue.get(), 'is (-10, (c, d))')
print('min_queue:', min_queue.get(), 'is (1, (a, b))')
print('min_queue:', min_queue.get(), 'is (2, (g, h))')
print('min_queue:', min_queue.get(), 'is (3, (e, f))')

print()

max_queue = PriorityQueue()
max_queue.put((-1, ('a', 'b')))
max_queue.put((10, ('c', 'd')))
max_queue.put((-3, ('e', 'f')))
max_queue.put((-2, ('g', 'h')))
print('max_queue:', max_queue.get(), 'is (-3, (e, f))')
print('max_queue:', max_queue.get(), 'is (-2, (g, h))')
print('max_queue:', max_queue.get(), 'is (-1, (a, b))')
print('max_queue:', max_queue.get(), 'is (10, (c, d))')

You could also subclass PriorityQueue:

from queue import PriorityQueue
from typing import Tuple, Any

class MinPriorityQueue(PriorityQueue):
    pass

class MaxPriorityQueue(PriorityQueue):
    def put(self, item: Tuple[int, Any], *args, **kwargs):
        return PriorityQueue.put(self, (item[0] * -1, item[1]), *args, **kwargs)

    def get(self, *args, **kwargs):
        priority, data = PriorityQueue.get(self, *args, **kwargs)
        return -1 * priority, data

max_queue = MaxPriorityQueue()
max_queue.put((1, ('a', 'b')))
max_queue.put((-10, ('c', 'd')))
max_queue.put((3, ('e', 'f')))
max_queue.put((2, ('g', 'h')))
print('max_queue:', max_queue.get(), 'is (3, (e, f))')
print('max_queue:', max_queue.get(), 'is (2, (g, h))')
print('max_queue:', max_queue.get(), 'is (1, (a, b))')
print('max_queue:', max_queue.get(), 'is (-10, (c, d))')

With updatable priorities

Python’s built-in PriorityQueue doesn’t support updating existing priorities.

import heapq

from queue import PriorityQueue

class UpdatablePriorityQueue(PriorityQueue):
    """PriorityQueue with updatable priorities."""

    def update(self, new_priority, item):
        """Update an item in the queue in O(n) time."""
        for i, it in enumerate(self.queue):
            if it[1] == item:
                self.queue[i] = (new_priority, item)
                heapq.heapify(self.queue)
                break

    def peek(self):
        return self.queue[0] if len(self.queue) > 0 else None

min_queue = UpdatablePriorityQueue()
min_queue.put((4, 'a'))
min_queue.put((3, 'b'))
print(min_queue.peek(), 'should be (3, b)')
min_queue.update(5, 'b')
print(min_queue.get(), 'should be (4, a)')
print(min_queue.get(), 'should be (5, b)')

Bibliography