The Project | About Us | Contribute | Donations | License

HOME

Every queue we have built so far shares one rule: the oldest comes out. But there are moments in TaskFlow when that rule is unfair in reverse: a priority 1 task ("the server is down") cannot wait behind twenty routine tasks that arrived earlier. We need a queue where dequeuing means "give me the most urgent", not "give me the oldest". That ADT is the priority queue, and it closes the second bridge from module 3: MinStack could query the minimum element in O(1), but not remove it; today we will learn to remove it. We will compare two honest implementations — an unsorted list and a sorted list, reusing insert_sorted from module 2 — understand their opposing costs, and use Python's professional tool, heapq, as a black box, paying special attention to a detail that separates correct code from treacherous code: ties and stability.

The priority queue ADT
Implementation 1: unsorted list
Implementation 2: sorted list (reusing insert_sorted)
Comparison table and the heap announcement
heapq: the professional black box
Ties and stability: the counter trick
TaskFlow: the urgent inbox

The priority queue ADT

The contract resembles the FIFO queue's, but the central promise changes:

Operation	FIFO queue	Priority queue
`enqueue(element)`	Enters at the back	Enters "wherever it belongs"
`dequeue()`	The oldest comes out	The most urgent comes out
`front()`	Inspects the oldest	Inspects the most urgent
`is_empty()` / `size()`	Same	Same

In TaskFlow, priority 1 is the highest, so "the most urgent" is the one with the lowest priority number: our priority queue is a min-priority queue. It is a common convention (think "priority 1" in tech support) and it fits Python's tools out of the box, since they also work with minimums.

Notice that the ADT says nothing about how this is achieved: it only promises that dequeue returns the minimum. As always since module 1, the same contract admits implementations with very different costs — and this time the tension between them is especially instructive: you can pay on the way in or pay on the way out, but with lists you pay.

Implementation 1: unsorted list

The lazy strategy: enqueuing means dropping the element at the end, in no particular order; dequeuing means finding the minimum and pulling it out.

class UnsortedPriorityQueue:
    """Enqueue O(1); dequeue O(n): searches for the minimum every time."""

    def __init__(self):
        self._items = []              # (priority, task) pairs, unordered

    def enqueue(self, priority, task):
        self._items.append((priority, task))           # O(1): at the end, done

    def dequeue(self):
        if self.is_empty():
            raise IndexError("dequeue from an empty queue")
        # Find the position of the minimum: full scan, O(n)
        best = 0
        for i in range(1, len(self._items)):
            if self._items[i][0] < self._items[best][0]:
                best = i
        return self._items.pop(best)[1]                # pop(i) is O(n) too

    def front(self):
        if self.is_empty():
            raise IndexError("front of an empty queue")
        return min(self._items, key=lambda pair: pair[0])[1]   # O(n)

    def is_empty(self):
        return len(self._items) == 0

    def size(self):
        return len(self._items)

Analysis: enqueue is O(1) — unbeatable — but dequeue scans the whole list to locate the minimum (O(n)) and then removes it with pop(best), which shifts the elements after it (O(n) again, an old acquaintance from module 1). When is it worth it? When you enqueue a great deal and dequeue very little: for example, when you accumulate thousands of candidates but will only extract a few.

Implementation 2: sorted list (reusing `insert_sorted`)

The far-sighted strategy: keep the collection always sorted by priority, so the minimum is ready at the front. When we wrote insert_sorted on the LinkedList back in module 2, we announced it was "a bridge toward priority queues". This is the moment to cross it: if the linked list is kept sorted ascending by priority, the most urgent element sits at the head, and removing it is the O(1) remove_front from lesson 04-02.

class SortedPriorityQueue:
    """Enqueue O(n) (sorted insertion); dequeue O(1) (the head)."""

    def __init__(self):
        self._items = LinkedList()      # sorted ascending by priority

    def enqueue(self, priority, task):
        # insert_sorted from module 2, storing (priority, task) pairs:
        # walks to the first node with a HIGHER priority and inserts before it.
        new = Node((priority, task))
        if (self._items.head is None
                or priority < self._items.head.data[0]):
            new.next = self._items.head                # new head
            self._items.head = new
            if new.next is None:
                self._items.tail = new
        else:
            current = self._items.head
            # Advance while the next node exists and is not less urgent.
            # NOTE the <=: ties go BEHIND the elements already there
            # (this preserves arrival order among equals: stability).
            while (current.next is not None
                   and current.next.data[0] <= priority):
                current = current.next
            new.next = current.next
            current.next = new
            if new.next is None:
                self._items.tail = new
        self._items.size += 1

    def dequeue(self):
        if self.is_empty():
            raise IndexError("dequeue from an empty queue")
        return self._items.remove_front()[1]             # O(1)

    def front(self):
        if self.is_empty():
            raise IndexError("front of an empty queue")
        return self._items.head.data[1]                  # O(1)

    def is_empty(self):
        return self._items.size == 0

    def size(self):
        return self._items.size

Now the costs flip: enqueue walks the list until it finds the gap (O(n)), but dequeue and front are O(1). When is it worth it? When you dequeue and inspect far more than you enqueue — for example, a dashboard that constantly asks "what is the next urgent item?".

A fine detail we left commented in the code: the <= in the loop makes a new task with the same priority as existing ones land behind them. Among equals, arrival order: we will keep this concept — stability — for the ties section, because with heapq we will have to earn it the hard way.

Comparison table and the heap announcement

Implementation	`enqueue`	`dequeue`	`front`	When to choose it?
Unsorted list	O(1)	O(n)	O(n)	Many insertions, few extractions
Sorted list (`insert_sorted`)	O(n)	O(1)	O(1)	Few insertions, many extractions/queries
Heap — module 6	O(log n)	O(log n)	O(1)	The general case: a mix of both

The table reveals a pattern you will see again and again in data structures: two symmetric solutions that pay O(n) on opposite operations, and a third, more sophisticated structure that balances: neither O(1) nor O(n), but O(log n) for both. To get a feel for what that means: with n = 1,000,000, O(n) is a million steps and O(log n) is about 20. That structure is the heap, a tree with a very clever ordering property... which is why it lives in module 6 (lesson 06-07), after we have learned about trees. We are not going to open it here: we are going to use it.

`heapq`: the professional black box

Python ships the heap in the standard heapq module. We will use it as a black box: we know what it promises (minimum out, O(log n) per operation) without yet looking at how it achieves it. It is exactly the ADT discipline from module 1 applied to a real library.

heapq does not define a class: it offers functions that operate on an ordinary list while maintaining the heap property on it:

import heapq

pending = []                          # a regular list will act as the heap

heapq.heappush(pending, 3)            # enqueue: O(log n)
heapq.heappush(pending, 1)
heapq.heappush(pending, 2)

print(pending[0])                     # 1 → the minimum is ALWAYS at [0] (front)
print(heapq.heappop(pending))         # 1 → dequeue: extracts the minimum, O(log n)
print(heapq.heappop(pending))         # 2
print(heapq.heappop(pending))         # 3

Black-box rules:

heappush(lst, element) = enqueue; heappop(lst) = dequeue (the minimum); lst[0] = front; len(lst) = size.
The list must only be touched through heapq functions (or read at [0]). An append or a sort on your own breaks the internal property and the minimum is no longer guaranteed.
A curiosity that connects to module 6: although the inside of pending is a seemingly unordered list (print it!), its structure encodes a tree. That an array can "be" a tree is one of the beautiful ideas we will unveil there.

What if the elements are not bare numbers? heapq compares elements against each other with <. With tuples, Python compares component by component, so the custom is to enqueue tuples whose first field is the priority: (priority, task). And here a serious trap appears.

Ties and stability: the counter trick

Let's try enqueuing tasks (our dicts) with their priority:

import heapq

urgent = []
t1 = {"id": 1, "title": "Restart server", "priority": 1, "status": "pending"}
t2 = {"id": 2, "title": "Notify clients", "priority": 1, "status": "pending"}

heapq.heappush(urgent, (t1["priority"], t1))
heapq.heappush(urgent, (t2["priority"], t2))   # TypeError!

TypeError: '<' not supported between instances of 'dict' and 'dict'. Why? Both tuples tie on the first field (1 == 1), so Python moves on to compare the second... and dicts do not know how to compare with <. The tie breaks the program.

And even if the elements were comparable, a subtler problem would remain: between two priority 1 tasks, which should come out first? The fair answer — and what any user expects — is the one that arrived first: equal priority, FIFO. That property is called stability, and heapq on its own does not guarantee it (the heap's internal order does not remember arrivals).

The canonical trick solves both problems at once: enqueue tuples with three fields, (priority, counter, task), where counter is an integer that grows with each insertion:

If the priorities tie, the counter is compared, and it never ties → the dicts are never compared (goodbye TypeError).
The lower counter corresponds to the earlier arrival → ties come out in arrival order (stability guaranteed).

import heapq
from itertools import count

urgent = []
counter = count()         # count() generates 0, 1, 2, ... a new one per next()

heapq.heappush(urgent, (t1["priority"], next(counter), t1))
heapq.heappush(urgent, (t2["priority"], next(counter), t2))   # now it works

priority, _, task = heapq.heappop(urgent)
print(task["title"])      # "Restart server": it arrived before its tie

This (priority, counter, element) pattern is so standard that you will find it verbatim in Python's official documentation and in production code. Memorize it the way you memorize an idiom.

TaskFlow: the urgent inbox

Let's package the pattern into this lesson's TaskFlow piece: the urgent inbox, where tasks from any source get dumped and the most urgent one is always served (remember: priority 1 = highest), with ties resolved by arrival:

import heapq
from itertools import count

class UrgentInbox:
    """Priority queue of TaskFlow tasks on top of heapq.

    dequeue() returns the task with the lowest priority number;
    at equal priority, the one enqueued first (stable).
    """

    def __init__(self):
        self._heap = []
        self._counter = count()

    def enqueue(self, task):
        entry = (task["priority"], next(self._counter), task)
        heapq.heappush(self._heap, entry)                 # O(log n)

    def dequeue(self):
        if self.is_empty():
            raise IndexError("dequeue from an empty inbox")
        return heapq.heappop(self._heap)[2]               # O(log n); [2] = task

    def front(self):
        if self.is_empty():
            raise IndexError("front of an empty inbox")
        return self._heap[0][2]                           # O(1)

    def is_empty(self):
        return len(self._heap) == 0

    def size(self):
        return len(self._heap)


# --- Usage ---
inbox = UrgentInbox()
inbox.enqueue({"id": 4, "title": "Update documentation",
               "priority": 3, "status": "pending"})
inbox.enqueue({"id": 5, "title": "Server down",
               "priority": 1, "status": "pending", "assigned_to": "anna"})
inbox.enqueue({"id": 6, "title": "Client locked out",
               "priority": 1, "status": "pending", "assigned_to": "bruno"})
inbox.enqueue({"id": 7, "title": "Review CSS styles",
               "priority": 2, "status": "pending"})

while not inbox.is_empty():
    task = inbox.dequeue()
    print(f"priority {task['priority']}: {task['title']}")

# priority 1: Server down              ← tie at 1: the earlier arrival comes out
# priority 1: Client locked out
# priority 2: Review CSS styles
# priority 3: Update documentation

The output sums up the whole lesson: priority rules, not arrival (task 5 overtakes task 4); and among ties, arrival rules (5 before 6). The heap does both in O(log n) per operation, and in module 6 we will finally open the box to see the tree that makes it possible.

Common Mistakes and Tips

Enqueuing (priority, task) without a counter: it works in tests... until the first tie between dicts, and then TypeError in production. Always use (priority, counter, element) when the element is not comparable (and even when it is: you gain stability).
Getting the direction of priority wrong: heapq extracts the minimum. With our convention (1 = highest) it fits directly; if your convention were "higher number = more urgent", you would have to enqueue (-priority, counter, task). Document the project's convention and never mix them.
Touching the heap list from outside: a lst.append(x) or lst.sort() on the heap silently breaks its internal invariant: it does not fail right away, it fails later, returning wrong minimums. The heap list is heapq's exclusive territory (which is why UrgentInbox hides it behind _).
Expecting reordering after changing the priority of an already-enqueued task: mutating task["priority"] does not relocate its entry in the heap. The professional pattern is to enqueue a new entry and mark the old one as invalidated (or rebuild the queue if it is small). We will work on this idea in the module 8 projects.
Tip: if you hesitate between the sorted list, the unsorted list, or heapq, count operations: do insertions dominate, do extractions, or are they even? This lesson's comparison table is literally a decision guide; keep it at hand.

Exercises

Exercise 1: predict with ties

Without running anything, state the order in which the titles come out if these tasks are enqueued into an UrgentInbox (in this order) and then everything is dequeued: ("Backup", 2), ("Fire", 1), ("Report", 2), ("Rescue", 1). Justify each position with the relevant rule (priority or stability).

Exercise 2: stable `dequeue` in the UNSORTED list

The UnsortedPriorityQueue from this lesson is not stable: on a tie, it returns the first one it finds, which is only correct by accident. Modify dequeue so the minimum search uses strict < (not <=) when comparing, and reason why that — given that append adds at the end — guarantees that among ties the oldest comes out.

Exercise 3: merging two inboxes

TaskFlow's "web" and "mobile" teams keep separate urgent inboxes and are merging into a single team. Write a function merge(inbox_a, inbox_b) that returns a new UrgentInbox with all the tasks from both, preserving the priority order. Is stability across inboxes preserved too? Reason your answer.

Solutions

Solution 1: Exit order: Fire, Rescue, Backup, Report.

Fire (priority 1): global minimum.
Rescue (priority 1): ties with Fire, but Fire arrived first (lower counter) → stability.
Backup (priority 2): nothing below 2 remains; ties with Report and arrived first.
Report (priority 2): last, by stability.

Solution 2:

def dequeue(self):
    if self.is_empty():
        raise IndexError("dequeue from an empty queue")
    best = 0
    for i in range(1, len(self._items)):
        if self._items[i][0] < self._items[best][0]:   # STRICT <
            best = i
    return self._items.pop(best)[1]

Reasoning: append places each new element behind the earlier ones, so inside the list ties sit in arrival order. The strict < search only switches candidates for a strictly better element; on a tie it keeps the one it already had, which — scanning left to right — is the one with the lower index = the oldest. With <= the opposite would happen: it would keep the last tie, reversing arrival order.

Solution 3:

def merge(inbox_a, inbox_b):
    result = UrgentInbox()
    for inbox in (inbox_a, inbox_b):
        while not inbox.is_empty():
            result.enqueue(inbox.dequeue())
    return result

The priority order is always preserved: each task is re-enqueued with its priority and the new heap will order them. Stability is preserved within each inbox (we dequeue in stable order and re-enqueue in that same order, with fresh increasing counters), but across inboxes there is no historical guarantee: every task from inbox_a receives lower counters than those from inbox_b, so in ties across inboxes inbox a always wins, not the task that actually arrived first in time. For a historically fair merge, a global timestamp would have had to be stored on each task. Note the improvement over naive versions: the function consumes the original inboxes; if you wanted to keep them, you would have to re-enqueue into the originals too (rotation) or expose a copy.

Conclusion

The priority queue changes the promise of dequeue: it is no longer the oldest that comes out, but the most urgent — in TaskFlow, the one with the lowest priority number. We saw the dilemma of the list implementations (unsorted: enqueue O(1) / dequeue O(n); sorted with insert_sorted: enqueue O(n) / dequeue O(1)), and we used the balanced solution from the standard library, heapq, with the (priority, counter, task) idiom that avoids the TypeError on ties and guarantees stability: at equal urgency, arrival order. The heap inside heapq — and why it achieves O(log n) — is an appointment we keep for lesson 06-07, once we know about trees. One bridge from module 3 remains to cross, the very first one: that BoundedHistory which needed efficiency at both ends at once. The structure that achieves it — the double-ended queue, or deque — and the standard library gem that implements it, collections.deque, await us in the next lesson.

Priority Queues

Contents

The priority queue ADT

Implementation 1: unsorted list

Implementation 2: sorted list (reusing `insert_sorted`)

Comparison table and the heap announcement

`heapq`: the professional black box

Ties and stability: the counter trick

TaskFlow: the urgent inbox

Common Mistakes and Tips

Exercises

Exercise 1: predict with ties

Exercise 2: stable `dequeue` in the UNSORTED list

Exercise 3: merging two inboxes

Solutions

Conclusion

Data Structures Course

Module 1: Introduction to Data Structures

Module 2: Lists

Module 3: Stacks

Module 4: Queues

Module 5: Hash Tables and Dictionaries

Module 6: Trees

Module 7: Graphs

Module 8: Conclusions and Further Resources

Priority Queues

Contents

The priority queue ADT

Implementation 1: unsorted list

Implementation 2: sorted list (reusing insert_sorted)

Comparison table and the heap announcement

heapq: the professional black box

Ties and stability: the counter trick

TaskFlow: the urgent inbox

Common Mistakes and Tips

Exercises

Exercise 1: predict with ties

Exercise 2: stable dequeue in the UNSORTED list

Exercise 3: merging two inboxes

Solutions

Conclusion

Data Structures Course

Module 1: Introduction to Data Structures

Module 2: Lists

Module 3: Stacks

Module 4: Queues

Module 5: Hash Tables and Dictionaries

Module 6: Trees

Module 7: Graphs

Module 8: Conclusions and Further Resources

Implementation 2: sorted list (reusing `insert_sorted`)

`heapq`: the professional black box

Exercise 2: stable `dequeue` in the UNSORTED list