Priority Queues, Sorted Lists and Summary Statistics

The deque from the previous article hands out jobs in the order they arrived. Sometimes an urgent job has to go first. A queue that always hands out the most urgent item next is a priority queue, and the heapq module builds one on a plain list.

import heapq

jobs: list[tuple[int, str]] = []
heapq.heappush(jobs, (2, "send report"))
heapq.heappush(jobs, (1, "fix outage"))
heapq.heappush(jobs, (3, "tidy logs"))
assert heapq.heappop(jobs) == (1, "fix outage")

The heappush and heappop functions keep the list in a shape called a min-heap. The smallest item is always at index 0, and the rest of the list is only partly sorted. Each push or pop takes O(log n) time. heapq.heapify(items) turns an existing list into a heap in O(n). The exercise uses heapq. The two smaller modules at the end, bisect and statistics, often come up next to it.

Break ties by arrival order

Python compares tuples one item at a time. So when two jobs have the same priority, the tuple (priority, name) compares their names. The jobs come out in alphabetical order, not in the order you pushed them. If the payloads are a type with no order, such as dicts, the comparison raises TypeError instead.

Add a sequence number that goes up by one on every push. Put it between the priority and the payload:

import heapq
from dataclasses import dataclass, field


@dataclass(order=True, slots=True)
class Entry:
    priority: int
    sequence: int
    name: str = field(compare=False)


heap = [Entry(2, 0, "zebra"), Entry(2, 1, "apple"), Entry(1, 2, "urgent")]
heapq.heapify(heap)
assert [heapq.heappop(heap).name for _ in range(3)] == ["urgent", "zebra", "apple"]

With order=True, the dataclass compares its fields in order, priority first and then sequence. field(compare=False) leaves name out of the comparison. It gives name no default value. A plain tuple (priority, sequence, name) works too, because two sequence numbers are never equal.

Check your understanding How can a queue cancel an entry without removing it from the middle of the heap? Show answer

Mark the entry as cancelled, and skip it when it reaches the top. This is called lazy deletion. One pop may throw away several cancelled entries before it finds a live one. Each cancelled entry takes up memory until a pop reaches it.

Check your understanding Is a set of cancelled task names enough to track cancellations? Show answer

Only if a name is never used again while an old entry for it can still be in the heap. Otherwise a cancelled old entry and a new live one share a name, and the queue mixes them up. If names can repeat, give each entry its own id and track the ids. The exercise promises that names are never reused, so a set of names is enough there.

Find positions in a sorted list with bisect

from bisect import bisect_left, bisect_right

values = [1, 2, 2, 4]
assert bisect_left(values, 2) == 1   # the first 2
assert bisect_right(values, 2) == 3  # just after the last 2
assert bisect_right(values, 2) - bisect_left(values, 2) == 2  # how many 2s

The bisect functions do a binary search, so they take O(log n) comparisons. Sort the list before you search it. To test whether target is in the list, set i = bisect_left(values, target) and check i < len(values) and values[i] == target.

bisect.insort finds the position and inserts the item there, so the list stays sorted. The search is fast, but inserting into a list shifts every later item, which takes O(n). Many inserts in a row cost O(n²) in total. If you need the sorted list only at the end, append everything and sort once.

Pick the summary with statistics

The rolling-median exercise used statistics.median. The module has a function for each common summary:

Function	Returns
`mean(data)`	The average
`median(data)`	The middle value, or the mean of the two middle values for an even count
`stdev(data)`, `variance(data)`	The spread of a sample, dividing by n - 1
`pstdev(data)`, `pvariance(data)`	The spread of a whole population, dividing by n
`quantiles(data, n=4)`	The three cut points that split the data into quarters

Use the sample versions when your data is a sample from something larger, and the population versions when it is the whole set. All of these functions raise StatisticsError on empty input, and stdev and variance need at least two values.

Going further

Python 3.14 added max-heap functions such as heapify_max, heappush_max and heappop_max. The exercises here run on Python 3.12, so they use the min-heap functions.

Since Python 3.10, the bisect functions accept key=. The key applies to the list items but not to the value you search for. So bisect_left(people, 30, key=get_age) searches with an age, not a person.