Skip to content

Getting started

Installation

pip install iheap

iheap supports CPython 3.11 and later. Every queue runs on the native Rust core; pip picks the right wheel for your interpreter:

Interpreter Wheel Synchronization
CPython 3.11+ (regular, with the GIL) cp311-abi3 none needed: the GIL is the lock
CPython 3.14t (free-threaded) cp314t a mutex per heap
CPython 3.15t and later (free-threaded) cp315-abi3t a mutex per heap
PyPy 3.11 pp311 the GIL

Native wheels cover:

  • Linux (glibc): x86_64, i686, aarch64, armv7, ppc64le, s390x;
  • Linux (musl): x86_64, i686, aarch64, armv7;
  • macOS: x86_64 and arm64;
  • Windows: x64, x86 and arm64.

Free-threaded and PyPy wheels are built for the 64-bit targets. On any other platform pip builds the extension from the source package, which needs Rust 1.88 or later. Free-threaded CPython 3.13 is not supported: the Rust bindings support free-threading from 3.14, where it stopped being experimental.

The basics

A PQDict maps keys to values. Each value is also the item's priority (see custom priorities for how to change that). The item with the smallest priority is always on top.

from iheap import PQDict

pq = PQDict({"write docs": 2, "fix bug": 1})
pq["release"] = 3          # insert
pq["fix bug"] = 5          # change a priority: O(log n)
del pq["release"]          # remove any item: O(log n)

assert pq.top() == "write docs"           # peek at the top key
assert pq.topitem() == ("write docs", 2)  # ... or the top item
assert pq.popitem() == ("write docs", 2)  # remove the top item
assert len(pq) == 1

To consume the queue in priority order, use the destructive iterators:

pq = PQDict({"c": 3, "a": 1, "b": 2})
assert list(pq.popkeys()) == ["a", "b", "c"]
assert len(pq) == 0

popvalues() and popitems() work the same way.

Max-queues

Pass reverse=True, or use maxpq, to put the largest priority on top:

from iheap import PQDict, maxpq, minpq

scores = maxpq(alice=90, bob=72, carol=85)
assert scores.top() == "alice"

assert PQDict({"x": 1, "y": 2}, reverse=True).top() == "y"
assert minpq(x=1, y=2).top() == "x"

Custom priorities

Pass key= to compute an item's priority from its key and value. The function is called as key(item_key, value). The value is stored unchanged; only the returned priority is sent to the native heap.

tasks = PQDict(key=lambda name, task: task["deadline"])
tasks["deploy"] = {"deadline": 17.5, "owner": "ops"}
tasks["review"] = {"deadline": 9.0, "owner": "dev"}

name, task = tasks.popitem()
assert name == "review" and task["owner"] == "dev"

The key can take part in the priority too, for example to break ties or to weight items:

# Shorter names first among equal deadlines.
jobs = PQDict(key=lambda name, deadline: deadline + len(name) / 1000)
jobs.update({"backup": 3.0, "db": 3.0, "etl": 1.0})
assert list(jobs.popkeys()) == ["etl", "db", "backup"]

Difference from pqdict

pqdict's key function receives only the value. To port such code, change key=f to key=lambda k, v: f(v). When a key function is set, replace_key and swap_priority recompute the affected priorities, because the key is an input. The same (key, value) function form is used by nsmallest and nlargest.

If you change a value in place in a way that changes its priority, tell the queue with heapify:

tasks["deploy"]["deadline"] = 1.0   # mutated in place: the heap doesn't know
tasks.heapify("deploy")             # re-prioritize one item in O(log n)
tasks.heapify()                     # ... or every item in O(n)

It is a full dict

Every dict operation works, and iteration follows insertion order, exactly like dict:

pq = PQDict({"b": 2, "a": 1})
pq.setdefault("c", 3)
assert pq.get("zzz", 0) == 0
assert list(pq) == ["b", "a", "c"]          # insertion order, not priority order
assert list(reversed(pq)) == ["c", "a", "b"]
assert pq == {"a": 1, "b": 2, "c": 3}       # compares equal to any mapping

merged = pq | {"d": 0}                      # returns a new PQDict
assert merged.top() == "d"
pq |= {"a": 10}
assert pq.top() == "b"

The one difference from dict

popitem() removes the top-priority item, not the last inserted one. pop() without arguments removes the top item and returns its key; pop(key[, default]) behaves exactly like dict.pop.

More priority-queue operations

pq = PQDict({"a": 5, "b": 7})

pq.additem("c", 6)                 # strict insert: KeyError if "c" exists
pq.updateitem("c", 8)              # strict update: KeyError if "c" is missing

# Push and pop in one pass (like heapq.heappushpop):
assert pq.pushpopitem("d", 1) == ("d", 1)    # would be on top: returned as-is
assert pq.pushpopitem("e", 6) == ("a", 5)    # "a" popped, "e" inserted

pq.replace_key("e", "E")           # rename a key, keeping value and priority
pq.swap_priority("b", "c")         # exchange two items' values in O(1)
assert pq == {"b": 8, "c": 7, "E": 6}

For one-off selection, nsmallest and nlargest run in native code:

from iheap import nlargest, nsmallest

prices = {"apple": 1.2, "pear": 0.8, "kiwi": 2.5, "fig": 3.1}
assert nsmallest(2, prices) == ["pear", "apple"]
assert nlargest(1, prices) == ["fig"]

Example: Dijkstra's shortest paths

Decrease-key is what makes an indexed heap the natural fit here:

from iheap import PQDict

def dijkstra(graph, source):
    dist = {}
    frontier = PQDict({source: 0.0})
    while frontier:
        node, d = frontier.popitem()
        dist[node] = d
        for neighbour, weight in graph[node].items():
            if neighbour in dist:
                continue
            candidate = d + weight
            if candidate < frontier.get(neighbour, float("inf")):
                frontier[neighbour] = candidate     # insert or decrease-key
    return dist

graph = {
    "a": {"b": 7, "c": 9, "f": 14},
    "b": {"a": 7, "c": 10, "d": 15},
    "c": {"a": 9, "b": 10, "d": 11, "f": 2},
    "d": {"b": 15, "c": 11, "e": 6},
    "e": {"d": 6, "f": 9},
    "f": {"a": 14, "c": 2, "e": 9},
}
assert dijkstra(graph, "a") == {"a": 0, "b": 7, "c": 9, "f": 11, "d": 20, "e": 20}

The low-level IndexedHeap

IndexedHeap is the native core. It stores only float priorities and gives back integer handles. Use it when you already have your own storage, such as numpy arrays or list-based graph nodes, and want to skip the PQDict layer, which costs roughly 100 ns per operation.

from iheap import IndexedHeap

heap = IndexedHeap()                       # IndexedHeap(reverse=True) for max
names = {}                                 # your own handle -> data storage

for name, priority in [("x", 3.0), ("y", 1.0), ("z", 2.0)]:
    names[heap.push(priority)] = name

handle, priority = heap.peek()
assert names[handle] == "y" and priority == 1.0

z = next(h for h, n in names.items() if n == "z")
heap.update(z, 0.5)                        # O(log n), returns the old priority
assert names[heap.pop()[0]] == "z"

handles = heap.push_many([9.0, 8.0])       # bulk insert, O(n) heapify if large
heap.update_many(handles, [0.1, 0.2])      # bulk update
assert [p for _, p in heap.pop_many(2)] == [0.1, 0.2]

A handle is valid until its entry is popped or removed. After that it is recycled by later pushes. Handles are always below handle_bound, so a list indexed by handle is the fastest way to store data alongside them.

Things to know

  • Priorities are numbers. Any value that float() accepts works (int, float, Decimal, Fraction, numpy scalars), except str. Priorities are stored as float, so integers are exact up to 2⁵³. NaN raises ValueError, and non-numbers raise TypeError; the queue is left unchanged in both cases.
  • Ties are popped in unspecified order. To get FIFO order among equal priorities, add a tiny increasing tie-breaker, such as priority + sequence_number * 1e-9 (choose a scale below your priorities' resolution).
  • Bulk operations are faster. PQDict(data) and pq.update(many_items) build in O(n) with one native call, which beats inserting in a loop.
  • Every operation is atomic. If an operation raises, whether from a bad priority or from a key function, the queue is unchanged.
  • Thread safety. IndexedHeap can be shared between threads: concurrent calls on one heap are serialized (by the GIL on regular builds; by a lock per heap on free-threaded Python). On free-threaded Python, importing iheap keeps the GIL disabled. Compound sequences, such as peek then update, still need your own lock to be atomic. PQDict is not safe for concurrent mutation: each of its operations takes several steps across Python and the heap. Guard a PQDict that several threads modify with a threading.Lock; that is required on free-threaded Python and recommended everywhere.
  • Pickling and copying work for queues and heaps; heaps restore exactly, including which handles will be reused next.