Skip to content
Hello Python
Data Structure1 Practice1 Interview

Priority Queue

Abstract queue that removes the highest-priority item, commonly implemented with a heap. This guide turns that contract into deliberate Python operations, interview invariants, and a runnable practice loop.

Recognize it when

Consider Priority Queue when the prompt's constraints and required operations match this shape: Abstract queue that removes the highest-priority item, commonly implemented with a heap.

Pybit studies a professional Priority Queue interview workspace with precise technical objects.
On this page · Mental Model

Checking your account…

Sign in to save your code and progress across devices. The lesson and problem statement remain public.

Priority Queue Code Labs

Mental Model

A priority queue is an interface: removal returns the pending item with the best priority, not the oldest item. A heap is the usual implementation. Separating the interface from the storage matters because priorities, tie rules, and updates belong to the queue contract rather than heap mechanics.

Priority Is the Removal Contract

Python’s heapq removes the smallest tuple. Decide whether a smaller number means higher priority, or negate a numeric field consistently for maximum priority. The root represents the next valid item to process; other pending items need not be globally sorted.

Encode Stable Composite Priorities

Tuple entries such as (priority, counter, item) establish a total order. The monotonic counter preserves insertion order among equal priorities and prevents Python from comparing arbitrary item objects as a final tie breaker.

Preserve Stable Priority Ties

Reference
import heapq

def schedule(tasks):
    heap = []
    for counter, (priority, label) in enumerate(tasks):
        heapq.heappush(heap, (priority, counter, label))
    return [heapq.heappop(heap)[2] for _ in range(len(heap))]
Practice

Implement schedule(tasks). Each task is (priority, label); return labels from smallest priority to largest while preserving input order for ties.

Public tests

  • Verify explicit tie ordering

Loading interactive editor.If it does not appear, the reference and starter code remain readable.Reload editor

The heapq priority-queue notes(opens in a new tab) describe this pattern. Choose the tuple fields deliberately; an accidental name or object field can silently create the wrong tie policy.

Skip Stale Entries

heapq has no efficient search-and-update operation. Push a new entry and record its current priority in a dictionary. When popping, discard entries whose priority no longer matches the current record. The invariant is that the first non-stale root is the best live item.

Discard Stale Priority Entries

Reference
import heapq

def updated_schedule(initial, updates):
    heap = []
    current = {}
    counter = 0
    for item, priority in [*initial, *updates]:
        current[item] = (priority, counter)
        heapq.heappush(heap, (priority, counter, item))
        counter += 1
    order = []
    while heap:
        priority, version, item = heapq.heappop(heap)
        if current.get(item) != (priority, version):
            continue
        order.append(item)
        del current[item]
    return order
Practice

Implement updated_schedule(initial, updates). Push every priority update lazily and return each item once using only its latest priority and stable update order.

Public tests

  • Verify stale versions are skipped

Loading interactive editor.If it does not appear, the reference and starter code remain readable.Reload editor

Lazy deletion may leave old entries in memory until they reach the root. Complexity counts every pushed version, although each stale version is eventually popped at most once.

Choose a Priority Queue or Heap Selection

Use a priority queue when pending work changes over time and repeatedly removing the best item drives the algorithm, as in Dijkstra or scheduling. Use a bounded heap for a static/streaming top-k summary, a FIFO queue for equal-priority breadth order, and sorting when one final complete order is enough.

Common Pitfalls

  • Omitting a tie counter can compare non-orderable task objects and raise TypeError.
  • Updating only the dictionary without pushing a new heap entry hides the improved priority.
  • Returning the first popped tuple without checking staleness processes obsolete work.
  • Negating priority on insertion but not when interpreting output reverses semantics.
  • Assuming one heap entry per logical item understates lazy-update memory.

Explain It in an Interview

Define whether smaller or larger is better and specify the tie rule. State what makes a heap entry live and why skipping stale roots is safe. Give O(log n) per pushed or popped entry and explain that lazy duplicates affect n. This is different from the bounded top-k heap, whose root is the weakest retained answer rather than the next task to execute.

Update a Bounded Heap(opens in a new tab) contrasts static candidate selection. Seat Reservation Manager(opens in a new tab) uses the priority-queue contract to repeatedly allocate the smallest available seat.

Python Version Note

Python 3.11+

The examples use standard Python containers and syntax available in the supported browser runtime; Priority Queue itself is taught as an interview abstraction.

Verify in Python docs(opens in a new tab)

Complexity & Invariants

Interview bounds depend on the stated representation, input model, and real Python operations.

OperationAverageWorstInterview note
Core Priority Queue workflowO(n log k + k log k)O(n log k + k log k)Streaming top-k selection needs only the current k best candidates, with the weakest candidate exposed at the heap root. After each value, the heap contains the largest min(k, processed_count) values seen so far.

Space

O(k) for the demonstrated Priority Queue workflow.

Assumptions

  • Heap construction can be O(n); individual push and pop restore heap order in logarithmic time.
  • The bound counts the operations in Update a Bounded Heap and does not hide Python slicing, sorting, or copying.

Invariants Worth Saying Aloud

  1. State the precise Priority Queue invariant before coding and preserve it after every update, traversal step, or recursive return.
  2. Streaming top-k selection needs only the current k best candidates, with the weakest candidate exposed at the heap root. This remains true after every accepted operation.
  3. After each value, the heap contains the largest min(k, processed_count) values seen so far. This remains true after every accepted operation.

When To Use Or Avoid Priority Queue

Use It When

  • Use Priority Queue when its representation directly supports the prompt's repeated query or update.
  • Use it when the invariant can be stated before coding and preserved after every operation.

Choose Another Tool When

  • Avoid it when a simpler Python sequence or mapping communicates the same contract with less state.
  • Avoid it when building the structure costs more time or space than the number of required queries can justify.

Common Pitfalls

These are the mistakes most likely to survive a happy-path example and fail a boundary case.

Leaving the core transition unfinished

Placeholder code cannot preserve the Priority Queue invariant or satisfy the public contract.

Prevent it: Implement the smallest complete transition and run the public tests before adding optimizations.

Avoid

def largest_k_sorted(values, k):
    pass

Use instead

import heapq

def largest_k_sorted(values, k):
    heap = []
    for value in values:
        if len(heap) < k:
            heapq.heappush(heap, value)
        elif k and value > heap[0]:
            heapq.heapreplace(heap, value)
    return sorted(heap)

Where you will hit this: Update a Bounded Heap(opens in a new tab)

Breaking the central invariant

Evicts whenever size reaches k instead of exceeds k, leaving at most k minus one retained values.

Prevent it: Keep this invariant visible while editing: State the precise Priority Queue invariant before coding and preserve it after every update, traversal step, or recursive return.

Avoid

import heapq

def largest_k_sorted(values, k):
    heap = []
    for value in values:
        heapq.heappush(heap, value)
        if len(heap) >= k:
            heapq.heappop(heap)
    return sorted(heap)

Use instead

import heapq

def largest_k_sorted(values, k):
    heap = []
    for value in values:
        if len(heap) < k:
            heapq.heappush(heap, value)
        elif k and value > heap[0]:
            heapq.heapreplace(heap, value)
    return sorted(heap)

Where you will hit this: Update a Bounded Heap(opens in a new tab)

Hiding Python work inside the loop

Handle empty state, duplicate values, aliasing, and serialization boundaries explicitly in tests.

Prevent it: Account for every slice, copy, sort, membership test, and container mutation before claiming O(n log k + k log k).

Avoid

import heapq

def largest_k_sorted(values, k):
    heap = []
    for value in values:
        heapq.heappush(heap, value)
        if len(heap) >= k:
            heapq.heappop(heap)
    return sorted(heap)

Use instead

import heapq

def largest_k_sorted(values, k):
    heap = []
    for value in values:
        if len(heap) < k:
            heapq.heappush(heap, value)
        elif k and value > heap[0]:
            heapq.heapreplace(heap, value)
    return sorted(heap)

Where you will hit this: Seat Reservation Manager(opens in a new tab)

Reviewed References

Official Python documentation supports language behavior. Canonical problem pages provide additional practice context.