---
title: "heapq em Python: fila de prioridade e top-N na prática"
url: "https://python.dev.br/blog/python-heapq-fila-prioridade/"
markdown_url: "https://python.dev.br/blog/python-heapq-fila-prioridade.MD"
description: "Aprenda heapq em Python para fila de prioridade, heappush, heappop, nlargest, nsmallest, heapify e padrões seguros com tuplas, dataclasses e threads."
date: "2026-08-18"
author: "Equipe Python Dev BR"
---

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

Aprenda heapq em Python para fila de prioridade, heappush, heappop, nlargest, nsmallest, heapify e padrões seguros com tuplas, dataclasses e threads.


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](/algoritmos/pilhas-filas-python/), [collections](/blog/collections-python-guia-completo/) e [threading com Queue](/blog/python-threading-threadpoolexecutor/), 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.

```python
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

| Função | O que faz | Complexidade típica |
| --- | --- | --- |
| `heappush(h, item)` | insere mantendo o invariante do heap | O(log n) |
| `heappop(h)` | remove e devolve o menor | O(log n) |
| `heapify(x)` | transforma lista in-place em heap | O(n) |
| `heappushpop(h, item)` | push + pop em um passo | O(log n) |
| `heapreplace(h, item)` | pop + push (lista nunca vazia) | O(log n) |
| `nlargest(k, iterable)` | k maiores | ~O(n log k) |
| `nsmallest(k, iterable)` | k menores | ~O(n log k) |
| `merge(*iterables)` | mescla iteráveis já ordenados | lazy |

`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”:

```python
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:

```python
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:

```python
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`](/blog/dataclasses-python-guia-completo/) e `order=True`, você pode ordenar só pelo campo desejado:

```python
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.

```python
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.

```python
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`](/blog/python-threading-threadpoolexecutor/), 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:

```python
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`](/blog/python-itertools-iteracao-elegante/) quando o pipeline for preguiçoso de ponta a ponta.

## heapq vs deque vs Queue vs PriorityQueue

| Necessidade | Ferramenta |
| --- | --- |
| FIFO simples, single-thread | `collections.deque` |
| LIFO / undo | `list` ou `deque` |
| Prioridade, single-thread | `heapq` |
| FIFO entre threads com bloqueio | `queue.Queue` |
| Prioridade entre threads com bloqueio | `queue.PriorityQueue` |
| Alta vazão / multiprocessamento | multiprocessing queues ou broker (Redis, SQS) |

`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:

```python
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

```python
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`](/blog/collections-python-guia-completo/) — filas e pilhas densas.
- [`threading` + `queue`](/blog/python-threading-threadpoolexecutor/) — workers bloqueantes.
- [`functools`](/blog/python-functools-lru-cache-partial-reduce/) — `total_ordering` e chaves auxiliares.
- [`itertools`](/blog/python-itertools-iteracao-elegante/) — pipelines lazy antes do top-N.
- [Algoritmos: pilhas e filas](/algoritmos/pilhas-filas-python/) — visão didática que já cita `heapq` de passagem.
- [Estruturas de dados](/blog/estruturas-de-dados-python/) — 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`](/blog/python-threading-threadpoolexecutor/). Se for modelar estados e códigos de domínio na mesma aplicação, feche com [`Enum` / `StrEnum`](/blog/python-enum-strenum-flag/). E se quiser acompanhar vagas que pedem exatamente essas estruturas em entrevistas e no dia a dia, veja o [radar de vagas Python](/vagas/).