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ção | Função | Complexidade Temporal |
|---|---|---|
| Inserir um item | heapq.heappush(heap, item) | O(log n) |
| Extrair o menor | heapq.heappop(heap) | O(log n) |
| Inserir e extrair | heapq.heappushpop(heap, item) | O(log n) |
| Extrair e inserir | heapq.heapreplace(heap, item) | O(log n) |
| Criar heap de lista | heapq.heapify(list) | O(n) |
| N menores itens | heapq.nsmallest(n, iterable) | O(n log k) |
| N maiores itens | heapq.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.8Quando usar nlargest/nsmallest vs sorted()
| Cenário | Melhor abordagem | Por quê |
|---|---|---|
| N é 1 | min() / max() | O(n), mais simples |
| N é pequeno em relação ao total | nlargest / nsmallest | O(n log k), menos memória |
| N está perto do comprimento total | sorted(iterable)[:n] | O(n log n), mas rápido na prática |
| Todos os itens precisam estar ordenados | sorted() | 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 1Fila 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 logsO 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.