graphlib.TopologicalSorter resolve um problema comum em builds, pipelines e migrações: ordenar tarefas respeitando dependências. A entrada é um mapeamento de cada nó para seus predecessores, não para seus sucessores.

from graphlib import CycleError, TopologicalSorter

dependencias = {
    "testar": {"instalar"},
    "empacotar": {"testar"},
    "publicar": {"empacotar"},
    "instalar": set(),
}

try:
    ordem = tuple(TopologicalSorter(dependencias).static_order())
    print(ordem)
except CycleError as erro:
    print("Ciclo detectado:", erro.args[1])

static_order() é a opção simples quando basta uma sequência. Se houver um ciclo, CycleError identifica o problema. A ordem exata entre nós independentes não deve ser tratada como contrato, pois mais de uma sequência pode ser válida.

Boas práticas

Para execução paralela, use prepare(), get_ready() e done() para liberar tarefas à medida que seus predecessores terminam. Mantenha nós hashable e valide dados externos antes de montar o grafo. Ciclos normalmente indicam configuração inválida e devem impedir a execução, não ser ignorados.

Para filas por prioridade, veja heapq em Python. Se as tarefas forem corrotinas relacionadas, asyncio.TaskGroup ajuda a controlar falhas e cancelamento.

A documentação oficial de graphlib, consultada em 22 de julho de 2026, detalha a API e suas garantias.

Modelagem correta

No mapeamento de entrada, cada chave é uma tarefa e seu conjunto contém os predecessores que precisam terminar antes dela. Para afirmar que compilar depende de gerar, escreva {"compilar": {"gerar"}}. Essa direção merece atenção porque outros formatos de grafo listam sucessores. Um predecessor omitido das chaves ainda vira um nó.

Use identificadores hashable e estáveis, como strings ou enums. Guarde comandos e metadados em outro dicionário. Assim é possível validar, testar e exibir o grafo sem executar efeitos.

from graphlib import TopologicalSorter

grafo = {
    "buscar": set(),
    "validar": {"buscar"},
    "transformar": {"validar"},
    "relatorio": {"validar"},
    "publicar": {"transformar", "relatorio"},
}
ordem = list(TopologicalSorter(grafo).static_order())
posicao = {no: i for i, no in enumerate(ordem)}
assert posicao["validar"] < posicao["transformar"]
assert posicao["relatorio"] < posicao["publicar"]

Teste relações obrigatórias, não uma sequência inteira. transformar e relatorio são independentes e podem trocar de posição.

Ciclos e falhas

Um ciclo significa que não existe ordem completa válida. Pode ser uma autorreferência ou uma cadeia longa que volta ao início. CycleError carrega detalhes em args, mas uma aplicação deve convertê-los em uma mensagem clara do domínio.

Depois de prepare() detectar um ciclo, nós independentes ainda podem ficar disponíveis. Isso ajuda ferramentas de diagnóstico, porém builds e migrações geralmente devem falhar antes de qualquer efeito. Executar apenas a parte possível pode deixar estado parcial.

Execução paralela

Para explorar tarefas independentes, use prepare(), get_ready() e done(). O coordenador envia o lote pronto a um executor. Só depois do sucesso informa a conclusão, liberando sucessores.

from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
from graphlib import TopologicalSorter

ts = TopologicalSorter(grafo)
ts.prepare()
futuros = {}
with ThreadPoolExecutor(max_workers=4) as executor:
    while ts.is_active():
        for nome in ts.get_ready():
            futuros[executor.submit(executar, nome)] = nome
        concluidos, _ = wait(futuros, return_when=FIRST_COMPLETED)
        for futuro in concluidos:
            nome = futuros.pop(futuro)
            futuro.result()
            ts.done(nome)

Chame done() uma vez e somente após sucesso. Se result() lançar, o exemplo interrompe sem liberar dependentes. Em produção, cancele trabalhos pendentes e registre quais tarefas executaram. Marcar uma falha como concluída fornece uma premissa falsa aos sucessores.

Ciclo de vida e validação

Adicione dependências antes de ordenar. Depois de prepare(), trate o grafo como fechado e crie outra instância quando a configuração mudar. Normalize entradas externas e não altere conjuntos enquanto o sorter os utiliza.

Dependências duplicadas não mudam o significado, mas uma aresta ausente pode liberar trabalho cedo demais. Valide referências, nomes vazios, etapas desabilitadas e dependências condicionais. Para entrada não confiável, limite também quantidade de nós e arestas.

Determinismo

Um DAG pode admitir várias ordens. Não transforme a ordem observada entre nós independentes em contrato. Se logs e testes exigem reprodução, normalize a entrada e ordene apenas cada lote retornado por get_ready(). Ordenar alfabeticamente o resultado final pode violar dependências. Mesmo com despacho estável, a conclusão real varia no paralelismo.

Limites da biblioteca

A ordenação percorre nós e arestas, com custo proporcional ao tamanho do grafo. Normalmente executar as tarefas custa muito mais. graphlib, entretanto, não é um agendador completo: não oferece persistência, repetição, timeout, prioridade, bloqueio de recursos nem coordenação distribuída. Ele informa o que está liberado; a aplicação executa, observa e recupera.

O padrão serve para ETL, módulos de build, inicialização de serviços, migrações e pré-requisitos. Prioridade ou preferência de ordem não são necessariamente dependências e devem ser modeladas separadamente.

Estratégia de testes

Cubra grafo vazio, nó isolado, cadeia, bifurcação, convergência e ciclo. Verifique que cada predecessor antecede seu dependente e que cada nó aparece uma vez. No modo incremental, confirme que done() libera exatamente os sucessores esperados. Teste falha de uma tarefa sem chamar done() e garanta que dependentes não executem.

Também registre duração, estado e erro por nó. Se etapas produzem efeitos, torne-as idempotentes ou defina compensação, pois uma falha pode ocorrer depois de outras concluírem.

Checklist

Confirme a direção das arestas, normalize identificadores e rejeite ciclos antes de efeitos irreversíveis. Não dependa da ordem entre nós independentes. Só marque sucesso real com done(). Defina cancelamento e tratamento de estado parcial. Quando o fluxo precisar sobreviver a reinícios ou coordenar máquinas, use TopologicalSorter apenas no planejamento e acrescente execução durável.