API Reference

Priority Queue

Simple Priority Queue.

class simpledsa.priority_queue.PriorityQueue(key_func: ~typing.Callable = <function PriorityQueue.<lambda>>)

Bases: Generic[T, P]

A flexible Priority Queue implementation supporting both min-heap and max-heap behavior.

Items can either carry their own priority or be assigned one explicitly.

Parameters:

key_func (Callable, optional) – Function to extract priority from items. Default is the identity function (item is its own priority). For max-heap behavior, use: priority_functions.reverse

extend(items: List[T]) → None

Add multiple items to the queue, using each item as its own priority.

Parameters:

items – Sequence of items to add

Examples

>>> pq = PriorityQueue()
>>> pq.extend([3, 1, 4])
>>> list(pq.pop_all())
[1, 3, 4]

extend_with_priority(items: List[Tuple[T, P]]) → None

Add multiple items with explicit priorities to the queue.

Parameters:

items – Sequence of (item, priority) tuples

Examples

>>> pq = PriorityQueue()
>>> pq.extend_with_priority([("task1", 2), ("task2", 1)])
>>> list(pq.pop_all())
['task2', 'task1']

classmethod from_items(items: ~typing.List[~simpledsa.priority_queue.T], key_func: ~typing.Callable = <function PriorityQueue.<lambda>>) → PriorityQueue[T, P]

Create a PriorityQueue from a list of items.

Parameters:

• items – List of items to add to the queue

• key_func – Optional priority function

Returns:

New PriorityQueue containing the items

classmethod from_items_with_priority(pairs: ~typing.List[~typing.Tuple[~simpledsa.priority_queue.T, ~simpledsa.priority_queue.P]], key_func: ~typing.Callable = <function PriorityQueue.<lambda>>) → PriorityQueue[T, P]

Create a PriorityQueue from a list of (item, priority) pairs.

Parameters:

• pairs – List of (item, priority) tuples

• key_func – Optional priority function

Returns:

New PriorityQueue containing the items

is_empty() → bool

Check if the priority queue is empty.

classmethod merge(queues: List[PriorityQueue[T, P]]) → PriorityQueue[T, P]

Merge multiple PriorityQueues into a new one.

Parameters:

queues – List of PriorityQueues to merge

Returns:

New PriorityQueue containing all items

Note

All queues must use the same key_func

peek() → T

Return the highest priority item without removing it.

Returns:

The item with the highest priority

Raises:

IndexError – If the queue is empty

peek_with_priority() → Tuple[T, P]

Return the highest priority item without removing it.

Returns:

The item with the highest priority

Raises:

IndexError – If the queue is empty

pop() → T

Remove and return the highest priority item.

Returns:

The item with the highest priority

Raises:

IndexError – If the queue is empty

pop_all() → Iterator[T]

Iterate through and remove all items in priority order.

Returns:

Iterator yielding items in priority order

pop_all_with_priority() → Iterator[Tuple[T, P]]

Iterate through and remove all items in priority order.

Returns:

Iterator yielding items in priority order

pop_with_priority() → Tuple[T, P]

Remove and return the highest priority item and its priority.

Returns:

The item with the highest priority and its priority

Raises:

IndexError – If the queue is empty

push(item: T, priority: P | None = None) → None

Add an item to the priority queue.

Parameters:

• item – The item to be added

• priority (optional) – External priority value. If not provided, the item’s own value will be used as priority

Priority Functions

Default functions to set up priority rules for the Priority Queue.

simpledsa.priority_functions.by_attr(attr_name: str) → Callable[[Any], Any]

Create a priority function using an attribute of items.

simpledsa.priority_functions.by_length(x: Any) → int

Priority function using length of items.

simpledsa.priority_functions.reverse(x: Any) → Any

Priority function for max heap behavior.