Skip to content

Python heapq: Filas de prioridade e operações de heap simplificadas

Updated on

Você precisa processar itens por prioridade -- tarefas em um agendador de trabalhos, eventos em uma simulação ou o caminho mais curto em um grafo. Uma lista ordenada parece a escolha óbvia, mas inserir em uma lista ordenada é O(n), o que se torna um gargalo ao processar milhões de itens. Reordenar toda a lista cada vez que adiciona um elemento é ainda pior com O(n log n).

O custo se acumula rapidamente. Um agendador de trabalhos que reordena sua fila a cada inserção desperdiça a maior parte do tempo em tarefas administrativas. Um algoritmo de busca de caminhos que varre uma lista completa para encontrar a distância mínima a cada passo transforma uma operação de tempo linear em uma quadrática.

O módulo heapq do Python fornece uma fila de prioridade baseada em heap que oferece inserção O(log n) e extração O(log n) do menor elemento. Também inclui funções otimizadas para encontrar os N maiores ou menores elementos de qualquer iterável sem ordenar toda a coleção. Este guia cobre cada função do heapq com exemplos práticos.

📚

O que é um Heap?

Um heap é uma árvore binária onde cada nó pai é menor ou igual aos seus filhos (min-heap). O heapq do Python implementa um min-heap usando uma lista regular, onde o menor elemento está sempre no índice 0.

As operações principais e suas complexidades temporais:

OperaçãoFunçãoComplexidade Temporal
Inserir um itemheapq.heappush(heap, item)O(log n)
Extrair o menorheapq.heappop(heap)O(log n)
Inserir e extrairheapq.heappushpop(heap, item)O(log n)
Extrair e inserirheapq.heapreplace(heap, item)O(log n)
Criar heap de listaheapq.heapify(list)O(n)
N menores itensheapq.nsmallest(n, iterable)O(n log k)
N maiores itensheapq.nlargest(n, iterable)O(n log k)

Operações básicas do Heap

Criar e usar um Heap

import heapq
 
# Start with an empty heap (just a regular list)
heap = []
 
# Push items
heapq.heappush(heap, 5)
heapq.heappush(heap, 1)
heapq.heappush(heap, 3)
heapq.heappush(heap, 2)
 
print(heap)        # [1, 2, 3, 5] (heap order, not sorted)
print(heap[0])     # 1 (smallest element, always at index 0)
 
# Pop smallest item
smallest = heapq.heappop(heap)
print(smallest)    # 1
print(heap)        # [2, 5, 3]

Converter uma lista em Heap

import heapq
 
data = [5, 3, 8, 1, 2, 7, 4, 6]
heapq.heapify(data)  # In-place, O(n)
print(data)  # [1, 2, 4, 3, 5, 7, 8, 6] (heap order)
 
# Pop elements in sorted order
sorted_data = []
while data:
    sorted_data.append(heapq.heappop(data))
print(sorted_data)  # [1, 2, 3, 4, 5, 6, 7, 8]

heappushpop e heapreplace

Essas operações combinadas são mais eficientes que chamadas separadas de push/pop:

import heapq
 
heap = [1, 3, 5, 7]
heapq.heapify(heap)
 
# heappushpop: push first, then pop smallest
result = heapq.heappushpop(heap, 2)
print(result)  # 1 (pushed 2, popped 1)
print(heap)    # [2, 3, 5, 7]
 
# heapreplace: pop smallest first, then push
result = heapq.heapreplace(heap, 10)
print(result)  # 2 (popped 2, pushed 10)
print(heap)    # [3, 7, 5, 10]

nlargest e nsmallest

Essas funções encontram os N maiores ou menores elementos sem ordenar toda a coleção.

import heapq
 
scores = [85, 92, 78, 95, 88, 72, 91, 83, 97, 69]
 
# Top 3 scores
top_3 = heapq.nlargest(3, scores)
print(top_3)  # [97, 95, 92]
 
# Bottom 3 scores
bottom_3 = heapq.nsmallest(3, scores)
print(bottom_3)  # [69, 72, 78]

Com uma função chave

Ambas aceitam um parâmetro key para comparação personalizada:

import heapq
 
students = [
    {'name': 'Alice', 'gpa': 3.9},
    {'name': 'Bob', 'gpa': 3.5},
    {'name': 'Charlie', 'gpa': 3.8},
    {'name': 'Diana', 'gpa': 3.95},
    {'name': 'Eve', 'gpa': 3.2},
]
 
top_students = heapq.nlargest(3, students, key=lambda s: s['gpa'])
for s in top_students:
    print(f"{s['name']}: {s['gpa']}")
# Diana: 3.95
# Alice: 3.9
# Charlie: 3.8

Quando usar nlargest/nsmallest vs sorted()

CenárioMelhor abordagemPor quê
N é 1min() / max()O(n), mais simples
N é pequeno em relação ao totalnlargest / nsmallestO(n log k), menos memória
N está perto do comprimento totalsorted(iterable)[:n]O(n log n), mas rápido na prática
Todos os itens precisam estar ordenadossorted()Ordenação completa é apropriada

Implementar um Max-Heap

O heapq do Python só fornece um min-heap. Para criar um max-heap, negue os valores:

import heapq
 
# Max-heap using negation
max_heap = []
for val in [5, 1, 3, 8, 2]:
    heapq.heappush(max_heap, -val)
 
# Pop largest items
while max_heap:
    print(-heapq.heappop(max_heap), end=' ')
# 8 5 3 2 1

Fila de prioridade com objetos personalizados

Para objetos que não são diretamente comparáveis, use tuplas (priority, counter, item):

import heapq
from dataclasses import dataclass
 
@dataclass
class Task:
    name: str
    description: str
 
# Use (priority, tie_breaker, task) tuples
task_queue = []
counter = 0
 
def add_task(priority, task):
    global counter
    heapq.heappush(task_queue, (priority, counter, task))
    counter += 1
 
def get_next_task():
    if task_queue:
        priority, _, task = heapq.heappop(task_queue)
        return task
    return None
 
add_task(3, Task("Low", "Clean up logs"))
add_task(1, Task("Critical", "Fix production bug"))
add_task(2, Task("Medium", "Update docs"))
add_task(1, Task("Critical", "Patch security hole"))
 
while task_queue:
    task = get_next_task()
    print(f"{task.name}: {task.description}")
 
# Critical: Fix production bug
# Critical: Patch security hole
# Medium: Update docs
# Low: Clean up logs

O contador serve como desempate para que itens com a mesma prioridade sejam retornados na ordem de inserção (FIFO).

Casos de uso práticos

Mesclar iteráveis ordenados

import heapq
 
list1 = [1, 4, 7, 10]
list2 = [2, 5, 8, 11]
list3 = [3, 6, 9, 12]
 
merged = list(heapq.merge(list1, list2, list3))
print(merged)  # [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]

heapq.merge() é eficiente em memória -- não carrega todas as listas na memória ao mesmo tempo.

Mediana em execução

import heapq
 
class RunningMedian:
    """Track the median of a stream of numbers."""
    def __init__(self):
        self.low = []   # max-heap (negated values)
        self.high = []  # min-heap
 
    def add(self, num):
        heapq.heappush(self.low, -num)
        heapq.heappush(self.high, -heapq.heappop(self.low))
        if len(self.high) > len(self.low):
            heapq.heappush(self.low, -heapq.heappop(self.high))
 
    def median(self):
        if len(self.low) > len(self.high):
            return -self.low[0]
        return (-self.low[0] + self.high[0]) / 2
 
rm = RunningMedian()
for num in [2, 1, 5, 7, 2, 0, 5]:
    rm.add(num)
    print(f"Added {num}, median = {rm.median()}")

Caminho mais curto de Dijkstra

import heapq
 
def dijkstra(graph, start):
    """Find shortest paths from start to all nodes."""
    distances = {node: float('inf') for node in graph}
    distances[start] = 0
    pq = [(0, start)]
 
    while pq:
        dist, node = heapq.heappop(pq)
        if dist > distances[node]:
            continue
        for neighbor, weight in graph[node]:
            new_dist = dist + weight
            if new_dist < distances[neighbor]:
                distances[neighbor] = new_dist
                heapq.heappush(pq, (new_dist, neighbor))
 
    return distances
 
graph = {
    'A': [('B', 1), ('C', 4)],
    'B': [('C', 2), ('D', 5)],
    'C': [('D', 1)],
    'D': [],
}
 
print(dijkstra(graph, 'A'))
# {'A': 0, 'B': 1, 'C': 3, 'D': 4}

Analisar desempenho do Heap no Jupyter

Ao comparar algoritmos baseados em heap com alternativas ou perfilar o desempenho de filas de prioridade, a experimentação interativa acelera o processo. RunCell (opens in a new tab) é um agente de IA para Jupyter que pode ajudá-lo a fazer benchmarks de operações de heap, visualizar características de desempenho e otimizar suas escolhas de estruturas de dados de forma interativa.

FAQ

O que é heapq em Python?

heapq é um módulo da biblioteca padrão que implementa um min-heap usando uma lista regular do Python. Fornece funções para inserir e extrair itens em tempo O(log n) enquanto mantém a propriedade do heap (menor elemento sempre no índice 0). É a forma padrão de implementar filas de prioridade em Python.

Como criar um max-heap em Python?

O heapq do Python só suporta min-heaps. Para simular um max-heap, negue os valores ao inserir e negue novamente ao extrair: heapq.heappush(heap, -value) para inserir, -heapq.heappop(heap) para obter o maior. Para objetos complexos, negue a prioridade em uma tupla: (-priority, counter, item).

Quando devo usar heapq vs sorted()?

Use heapq.nlargest(n, data) ou heapq.nsmallest(n, data) quando precisar apenas dos top N itens e N for pequeno em relação ao tamanho total dos dados. Use sorted() quando precisar de todos os itens em ordem ou quando N estiver perto do tamanho total. Use min()/max() quando N for 1.

heapq é thread-safe?

Não. Chamadas individuais de heappush e heappop não são atômicas. Se múltiplas threads acessarem o mesmo heap, você precisa de sincronização externa. Para filas de prioridade thread-safe, use queue.PriorityQueue, que encapsula heapq com travamento adequado.

Qual é a complexidade temporal das operações do heapq?

heappush e heappop são ambos O(log n). heapify converte uma lista em heap em O(n). nlargest(k, data) e nsmallest(k, data) executam em O(n log k) onde n é o tamanho dos dados e k é o número de itens solicitados.

Conclusão

O módulo heapq do Python é a ferramenta padrão para filas de prioridade e seleção top-N. Sua abordagem baseada em heap fornece operações de push e pop em O(log n), tornando-o dramaticamente mais rápido que manter uma lista ordenada. Use heappush/heappop para padrões de filas de prioridade, nlargest/nsmallest para seleção top-N, merge para combinar sequências ordenadas e heapify quando tiver uma lista existente para converter.

📚