heapq no Python: filas de prioridade

Quando um programa precisa processar sempre o item mais urgente, menor ou mais próximo, ordenar a coleção inteira depois de cada alteração costuma ser desperdício. Sistemas de agendamento, simuladores, filas de tarefas, algoritmos de caminhos e análises de grandes conjuntos de dados frequentemente precisam apenas descobrir rapidamente qual elemento deve sair primeiro. O módulo heapq no Python oferece uma estrutura compacta para esse problema: o heap binário.

Neste guia, você aprenderá a criar filas de prioridade, inserir e remover itens, transformar listas existentes, selecionar os menores ou maiores valores e lidar com prioridades repetidas. O assunto complementa os artigos sobre estruturas do módulo collections, ordenação com sort e sorted, listas em Python, funções em Python e análise de desempenho com cProfile.

O que é um heap?

Um heap é uma árvore binária armazenada dentro de uma lista. No min-heap usado por padrão no Python, cada nó possui valor menor ou igual ao de seus filhos. Essa regra é chamada de propriedade ou invariante do heap. Ela não significa que a lista inteira esteja ordenada. Significa apenas que o menor elemento sempre ocupa a posição zero.

Para um índice k, os filhos aparecem em 2*k + 1 e 2*k + 2. O formato em lista evita criar objetos de árvore e aproveita memória contígua. A documentação oficial do heapq descreve essa representação, as operações disponíveis e as APIs de heap mínimo e máximo.

Por que usar heapq em vez de ordenar tudo?

Ordenar n elementos custa, em geral, O(n log n). Quando você adiciona uma tarefa e precisa somente do menor item, ordenar novamente toda a lista repete trabalho desnecessário. Em um heap, inserir ou remover a raiz custa O(log n), enquanto consultar o menor valor em heap[0] custa O(1).

Transformar uma lista inteira em heap com heapify() é uma operação linear, O(n). Por isso, se os valores já estão disponíveis, normalmente é melhor aplicar heapify uma vez do que inserir cada elemento individualmente.

Transformando uma lista com heapify

import heapq

numeros = [18, 4, 12, 7, 2, 30, 9]
heapq.heapify(numeros)

print(numeros)
print(numeros[0])  # menor valor

A ordem visual da lista pode parecer estranha, mas a propriedade do heap está preservada. Não compare a lista resultante com uma sequência totalmente ordenada. O contrato importante é que a raiz contém o menor item e que as operações do módulo mantêm a estrutura válida.

Inserindo e removendo elementos

Use heappush() para inserir e heappop() para remover o menor item:

import heapq

fila = []
heapq.heappush(fila, 20)
heapq.heappush(fila, 5)
heapq.heappush(fila, 12)

while fila:
    proximo = heapq.heappop(fila)
    print(proximo)

Os valores serão impressos em ordem crescente. Se você apenas precisa consultar a prioridade atual sem remover, leia fila[0]. Tanto heappop() quanto o acesso direto falham quando a lista está vazia. Verifique a condição antes da operação ou trate IndexError de forma explícita.

Criando uma fila de prioridade com tuplas

Na prática, uma fila normalmente guarda um valor de prioridade e a tarefa associada. Como tuplas são comparadas elemento por elemento, basta colocar a prioridade primeiro:

import heapq

fila = []
heapq.heappush(fila, (3, "gerar relatório"))
heapq.heappush(fila, (1, "corrigir indisponibilidade"))
heapq.heappush(fila, (2, "responder cliente"))

prioridade, tarefa = heapq.heappop(fila)
print(prioridade, tarefa)

Nesse exemplo, números menores representam maior urgência. Você pode inverter essa convenção, mas deve mantê-la consistente em todo o sistema.

Como tratar prioridades iguais

Se duas prioridades forem iguais, o Python tentará comparar o segundo elemento da tupla. Isso funciona para textos, mas pode falhar quando as tarefas são objetos sem ordenação definida. Além disso, talvez você queira preservar a ordem de chegada. Um contador crescente resolve os dois problemas:

import heapq
from itertools import count

contador = count()
fila = []

heapq.heappush(fila, (2, next(contador), {"id": 101}))
heapq.heappush(fila, (2, next(contador), {"id": 102}))

prioridade, ordem, tarefa = heapq.heappop(fila)
print(tarefa)

O contador atua como desempate estável. Como cada número é único, os dicionários nunca são comparados diretamente.

Usando dataclass para itens priorizados

Uma classe pode deixar o formato mais explícito. O campo da tarefa deve ser ignorado na comparação:

from dataclasses import dataclass, field
from typing import Any
import heapq

@dataclass(order=True)
class ItemPriorizado:
    prioridade: int
    item: Any = field(compare=False)

fila = [
    ItemPriorizado(4, "backup"),
    ItemPriorizado(1, "alerta crítico"),
]
heapq.heapify(fila)
print(heapq.heappop(fila).item)

Esse padrão melhora a legibilidade quando a fila circula por várias camadas da aplicação.

Atualizando ou removendo uma tarefa

Alterar um item no meio da lista pode quebrar o heap. Procurar a tarefa também custa O(n). Uma estratégia comum é manter um dicionário com as entradas atuais, marcar a versão antiga como removida e inserir uma nova entrada. Ao retirar elementos, o código ignora os marcados.

import heapq
from itertools import count

REMOVIDA = object()
fila = []
entradas = {}
contador = count()

def adicionar(tarefa, prioridade):
    if tarefa in entradas:
        remover(tarefa)
    entrada = [prioridade, next(contador), tarefa]
    entradas[tarefa] = entrada
    heapq.heappush(fila, entrada)

def remover(tarefa):
    entrada = entradas.pop(tarefa)
    entrada[2] = REMOVIDA

def retirar():
    while fila:
        prioridade, _, tarefa = heapq.heappop(fila)
        if tarefa is not REMOVIDA:
            del entradas[tarefa]
            return tarefa, prioridade
    raise KeyError("fila vazia")

Essa remoção preguiçosa mantém as operações principais eficientes e evita reconstruir o heap após cada mudança.

heappushpop e heapreplace

As funções combinadas são úteis quando o heap possui tamanho fixo. heappushpop(heap, item) insere o novo valor e remove o menor em uma operação otimizada. Ela devolve o menor entre o item novo e a raiz anterior, deixando o maior no heap.

heapreplace(heap, item) remove primeiro a raiz existente e depois insere o novo item. O tamanho permanece igual, mas a lista não pode estar vazia. A diferença importa quando você mantém os três maiores valores observados:

import heapq

top = [8, 2, 15]
heapq.heapify(top)

for valor in [3, 21, 7]:
    if valor > top[0]:
        heapq.heapreplace(top, valor)

print(sorted(top, reverse=True))

O heap guarda somente três elementos. A raiz representa o menor valor entre os atuais candidatos e pode ser substituída quando aparece um número melhor.

Encontrando os maiores e menores valores

nlargest() e nsmallest() retornam uma quantidade limitada de elementos e aceitam uma função key:

import heapq

produtos = [
    {"nome": "A", "preco": 80},
    {"nome": "B", "preco": 25},
    {"nome": "C", "preco": 110},
    {"nome": "D", "preco": 45},
]

mais_caros = heapq.nlargest(2, produtos, key=lambda p: p["preco"])
mais_barato = heapq.nsmallest(1, produtos, key=lambda p: p["preco"])
print(mais_caros)
print(mais_barato)

Essas funções são indicadas quando n é pequeno em relação ao conjunto. Para muitos resultados, sorted() pode ser mais eficiente. Quando você quer somente um item, use min() ou max().

Heap máximo no Python 3.14

Durante muitos anos, o padrão para simular um max-heap era inserir números negativos. O Python 3.14 adicionou funções explícitas: heapify_max(), heappush_max(), heappop_max(), heappushpop_max() e heapreplace_max().

import heapq

valores = [4, 19, 7, 12]
heapq.heapify_max(valores)
print(heapq.heappop_max(valores))  # 19

Se o projeto precisa suportar versões anteriores, a técnica de negar números ainda funciona, mas deve ser documentada. As APIs com sufixo _max são mais claras e evitam erros de sinal.

Mesclando sequências ordenadas

heapq.merge() combina várias entradas já ordenadas e devolve um iterador. Diferentemente de concatenar tudo e chamar sorted, ele não carrega necessariamente todos os dados na memória:

import heapq

log_a = [1, 4, 8]
log_b = [2, 3, 10]

for valor in heapq.merge(log_a, log_b):
    print(valor)

O método é útil para arquivos de log, resultados paginados e fluxos temporais. As entradas precisam estar ordenadas no mesmo sentido.

heapq ou queue.PriorityQueue?

heapq é um conjunto de funções para listas e não oferece bloqueios de sincronização. Quando várias threads produzem e consomem tarefas, a classe descrita na documentação de queue.PriorityQueue oferece operações protegidas, espera e limites de capacidade.

Use heapq em algoritmos locais, dentro de uma única thread ou quando você controla a sincronização. Use PriorityQueue quando precisa de uma fila concorrente pronta. Em aplicações assíncronas, considere também asyncio.PriorityQueue.

Exemplo de agendamento

Uma agenda pode ordenar tarefas pelo instante previsto. A data mais próxima sempre sai primeiro:

import heapq
from datetime import datetime, timedelta

agenda = []
agora = datetime.now()

heapq.heappush(agenda, (agora + timedelta(minutes=10), "enviar relatório"))
heapq.heappush(agenda, (agora + timedelta(minutes=2), "atualizar cache"))
heapq.heappush(agenda, (agora + timedelta(minutes=5), "verificar importações"))

momento, tarefa = heapq.heappop(agenda)
print(momento, tarefa)

Um agendador de produção também precisa de persistência, fuso horário, tentativas, cancelamento e proteção contra trabalhadores duplicados. O heap resolve a ordenação, não todo o ciclo de execução.

Erros comuns

• Imaginar que toda a lista está ordenada porque heap[0] é o menor item.
• Adicionar com append() em vez de heappush() depois que a lista virou heap.
• Remover posições arbitrárias e quebrar a invariante.
• Comparar tarefas incompatíveis quando duas prioridades são iguais.
• Usar números negativos sem documentar que o objetivo é um max-heap.
• Aplicar nlargest a quase todos os elementos quando sorted seria mais simples.
• Compartilhar a lista entre threads sem proteção.

Como testar uma fila de prioridade

Não teste a forma interna exata da lista, porque diferentes heaps válidos podem representar os mesmos dados. Teste o comportamento observável: sequência de remoção, desempate, atualização, remoção lógica e exceções.

def test_ordem_da_fila():
    fila = []
    heapq.heappush(fila, (3, "baixa"))
    heapq.heappush(fila, (1, "alta"))
    heapq.heappush(fila, (2, "media"))

    assert heapq.heappop(fila)[1] == "alta"
    assert heapq.heappop(fila)[1] == "media"
    assert heapq.heappop(fila)[1] == "baixa"

Boas práticas

• Defina claramente se menor número significa maior prioridade.
• Use contador para preservar a ordem de chegada.
• Centralize as operações em uma classe quando houver atualização e remoção.
• Use heapify() para coleções já existentes.
• Prefira funções combinadas em heaps de tamanho fixo.
• Meça antes de substituir uma ordenação simples.
• Documente a versão mínima do Python ao usar APIs de max-heap.

Conclusão

O heapq no Python permite acessar e remover prioridades sem ordenar a coleção inteira após cada alteração. Com heapify, heappush, heappop, funções combinadas, seleção dos maiores e menores itens e APIs de max-heap, o módulo cobre desde algoritmos pequenos até filas de tarefas robustas.

O ponto principal é respeitar a invariante do heap e desenhar corretamente o formato dos itens. Para casos simples, uma tupla de prioridade e tarefa basta. Para prioridades iguais, atualizações e cancelamentos, use contador, dicionário auxiliar e remoção preguiçosa. Assim, a fila permanece eficiente, previsível e fácil de testar.