heapq em Python: fila de prioridade e top-N na prática

O módulo heapq é a forma canônica de montar uma fila de prioridade em Python sem instalar nada. Em cima de uma lista comum ele mantém um binary heap: inserir e remover o menor elemento custam O(log n), e o menor item fica sempre em fila[0].

A recomendação direta é: use heappush / heappop para agendas, jobs, retries e dijkstra-like; use heapify quando a lista já existe; use nlargest / nsmallest para top-N sem ordenar tudo; e guarde itens como (prioridade, desempate, payload) para evitar TypeError em empates. Para a visão estrutural de pilhas e filas, collections e threading com Queue, este guia fecha o buraco do heap em si.

Neste tutorial você vai montar uma fila de prioridade real, extrair top-N de métricas, agendar tarefas com atraso, entender o min-heap e encaixar heapq em automações e backends brasileiros.

Por que não bastam sort e min?

Três situações clássicas:

1. Agenda contínua — chegam milhares de jobs; a cada tick você precisa só do próximo, não da lista toda ordenada.
2. Top-N em stream — quer os 10 maiores tickets de um mês com 2 milhões de linhas, sem materializar sorted(...)[-10:] em memória cara.
3. Custo logarítmico — list.sort() é O(n log n) sempre; heappush/heappop são O(log n) por operação.

import heapq

fila = []
heapq.heappush(fila, (2, "gerar-nf"))
heapq.heappush(fila, (1, "cobrar-pix"))
heapq.heappush(fila, (3, "enviar-email"))

print(heapq.heappop(fila))  # (1, 'cobrar-pix') — menor prioridade primeiro
print(fila[0])              # próximo sem remover: (2, 'gerar-nf')

Regra mental: menor número = maior urgência no min-heap padrão.

API essencial do heapq

fila[0] é sempre o menor depois de qualquer operação válida — mas fatiar ou indexar o meio da lista não devolve ordem crescente completa. Para varrer em ordem, vá dando heappop ou use nsmallest(len(h), h).

Min-heap, max-heap e prioridades invertidas

heapq só implementa min-heap. Para “maior prioridade numérica sai primeiro”:

import heapq

fila = []
# urgência 10 deve sair antes de urgência 1
heapq.heappush(fila, (-10, "cliente-vip"))
heapq.heappush(fila, (-1, "newsletter"))
heapq.heappush(fila, (-5, "cobranca"))

print(heapq.heappop(fila)[1])  # cliente-vip

O sinal negativo transforma o máximo em mínimo. Documente isso no código: times confundem 1 = alta com 1 = baixa o tempo todo.

O padrão seguro: (prioridade, contador, item)

Empurrar (prioridade, dict) ou (prioridade, objeto_sem_lt) quebra quando duas prioridades empatam:

import heapq

fila = []
heapq.heappush(fila, (1, {"id": "a"}))
heapq.heappush(fila, (1, {"id": "b"}))  # TypeError: '<' not supported between dicts

Corrija com um contador monotônico:

import heapq
from itertools import count

fila = []
seq = count()

def agendar(prioridade, payload):
    heapq.heappush(fila, (prioridade, next(seq), payload))

agendar(1, {"id": "a"})
agendar(1, {"id": "b"})
agendar(0, {"id": "urgente"})

while fila:
    prioridade, _, payload = heapq.heappop(fila)
    print(prioridade, payload["id"])

O desempate fica estável (FIFO entre iguais) e o heap nunca tenta comparar os dicionários.

Com dataclasses e order=True, você pode ordenar só pelo campo desejado:

from dataclasses import dataclass, field

@dataclass(order=True)
class Tarefa:
    prioridade: int
    nome: str = field(compare=False)

Ainda assim, o contador explícito é mais previsível em sistemas longos.

Top-N sem ordenar a coleção inteira

Caso típico de analytics brasileiro: ranking de SKUs, tickets de suporte, filiais por GMV.

import heapq

vendas = [
    ("SP-01", 182_340.55),
    ("RJ-02", 97_110.00),
    ("MG-01", 210_004.12),
    ("RS-03", 54_220.80),
    ("BA-01", 121_900.40),
]

top3 = heapq.nlargest(3, vendas, key=lambda x: x[1])
piores = heapq.nsmallest(2, vendas, key=lambda x: x[1])

print(top3)
print(piores)

Para um stream (gerador, cursor de banco, arquivo CSV grande), nlargest/nsmallest mantêm um heap de tamanho k — bem mais leve que carregar tudo e chamar sorted.

Quando k = 1, use max/min. Quando k ≈ n, sorted fica mais simples e claro.

Agenda de tarefas com timestamp

Filas de prioridade brilham em schedulers: retries de webhook, envio de boleto, lembretes, outbox.

import heapq
import time
from dataclasses import dataclass
from typing import Callable

@dataclass(order=True)
class Evento:
    quando: float
    nome: str
    acao: Callable[[], None]

agenda: list[Evento] = []

def marcar(atraso_s: float, nome: str, acao: Callable[[], None]) -> None:
    heapq.heappush(agenda, Evento(time.monotonic() + atraso_s, nome, acao))

def rodar_ate(vazio_timeout: float = 0.0) -> None:
    fim = time.monotonic() + vazio_timeout
    while agenda:
        agora = time.monotonic()
        if agenda[0].quando > agora:
            time.sleep(min(agenda[0].quando - agora, 0.05))
            if vazio_timeout and time.monotonic() > fim and agenda[0].quando > time.monotonic():
                break
            continue
        evento = heapq.heappop(agenda)
        evento.acao()

marcar(0.01, "pix-confirmado", lambda: print("baixar pedido"))
marcar(0.02, "nf-e", lambda: print("emitir NF-e"))
rodar_ate(1.0)

Em produção real você trocaria o sleep por um worker com threading, Celery/Redis ou um loop asyncio — a estrutura de dados continua sendo o heap.

merge: vários fluxos já ordenados

heapq.merge mescla iteráveis já ordenados de forma lazy — útil para logs rotacionados, partições diárias ou shards:

import heapq

manha = ["2026-08-18T08:01", "2026-08-18T08:15"]
tarde = ["2026-08-18T13:02", "2026-08-18T16:40"]
noite = ["2026-08-18T20:11"]

for ts in heapq.merge(manha, tarde, noite):
    print(ts)

Cada fonte precisa estar ordenada; o merge só garante a ordem global se essa pré-condição valer. Combine com itertools quando o pipeline for preguiçoso de ponta a ponta.

heapq vs deque vs Queue vs PriorityQueue

queue.PriorityQueue usa heapq por baixo e adiciona locks. Não misture heappush manual na mesma lista que uma PriorityQueue expõe — respeite a API de cada uma.

Trecho típico com threads:

from queue import PriorityQueue
from threading import Thread

pq: PriorityQueue[tuple[int, str]] = PriorityQueue()
pq.put((1, "reindexar"))
pq.put((0, "pagar-fornecedor"))

def worker() -> None:
    prioridade, nome = pq.get()
    print(nome)
    pq.task_done()

t = Thread(target=worker)
t.start()
pq.join()
t.join()

Armadilhas que quebram produção

1. Mutar um item já dentro do heap — mudar a prioridade “no lugar” corrompe o invariante. Remoção preguiçosa (marcar cancelado e ignorar no pop) é o padrão prático; heap binário clássico não tem decrease-key barato na stdlib.
2. Comparar payloads incomparáveis — sempre use o padrão (prio, seq, item).
3. Achar que list(fila) está ordenada — não está. Só a raiz é garantida.
4. Usar heapq como fila FIFO — empates dependem do desempate; para FIFO puro, use deque.
5. Esquecer que heappop em lista vazia levanta IndexError — verifique if fila: ou capture a exceção na borda do worker.
6. Misturar float("nan") como prioridade — NaN quebra ordenação. Valide entradas numéricas, especialmente lendo CSV brasileiro com células vazias.

Exemplo completo: fila de suporte com SLA

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

_seq = count()

@dataclass(order=True)
class Ticket:
    prioridade: int
    sequencia: int = field(compare=True)
    cliente: str = field(compare=False)
    assunto: str = field(compare=False)

class FilaSuporte:
    def __init__(self) -> None:
        self._heap: list[Ticket] = []

    def abrir(self, prioridade: int, cliente: str, assunto: str) -> None:
        heapq.heappush(
            self._heap,
            Ticket(prioridade, next(_seq), cliente, assunto),
        )

    def atender(self) -> Ticket | None:
        if not self._heap:
            return None
        return heapq.heappop(self._heap)

    def peek(self) -> Ticket | None:
        return self._heap[0] if self._heap else None

fila = FilaSuporte()
fila.abrir(2, "MEI-SP", "boleto vencido")
fila.abrir(0, "Enterprise-RJ", "API fora")
fila.abrir(1, "Ecommerce-MG", "PIX duplicado")

atual = fila.atender()
assert atual is not None
print(atual.cliente, atual.assunto)  # Enterprise-RJ API fora

Prioridade 0 atende primeiro; o contador garante ordem de chegada entre tickets iguais. Esse esqueleto escala para bots de Telegram, painéis internos e workers que leem de Postgres — a política de prioridade fica em um só lugar.

Quando não usar heapq

• Você precisa de fila FIFO pura → deque.
• Precisa buscar/remover item arbitrário com frequência → dict + estrutura auxiliar, ou biblioteca de heap indexado.
• Ordenação total one-shot → sorted.
• Distribuição entre máquinas → fila externa (SQS, RabbitMQ, Redis), não um heap in-process.
• Tempo real com deadlines rígidos de SO → APIs do sistema ou schedulers dedicados.

Relação com o restante da stdlib

• collections.deque — filas e pilhas densas.
• threading + queue — workers bloqueantes.
• functools — total_ordering e chaves auxiliares.
• itertools — pipelines lazy antes do top-N.
• Algoritmos: pilhas e filas — visão didática que já cita heapq de passagem.
• Estruturas de dados — mapa mental do ecossistema.

Conclusão

heapq resolve prioridade com custo logarítmico em cima de uma lista comum: agenda de jobs, top-N, merge de streams ordenados e filas de suporte com SLA. Lembre os três hábitos que evitam 90% dos bugs: (1) trate-o como min-heap; (2) use (prioridade, contador, item); (3) não confunda a lista do heap com uma sequência totalmente ordenada.

Se o próximo passo for concorrência, avance para ThreadPoolExecutor e Queue. Se for modelar estados e códigos de domínio na mesma aplicação, feche com Enum / StrEnum. E se quiser acompanhar vagas que pedem exatamente essas estruturas em entrevistas e no dia a dia, veja o radar de vagas Python.