O módulo graphlib oferece uma implementação da classe TopologicalSorter, usada para ordenar tarefas que possuem dependências. Uma ordenação topológica organiza os nós de um grafo direcionado acíclico de modo que cada dependência apareça antes do item que depende dela. Esse padrão aparece em sistemas de build, pipelines de dados, migrações, instalação de pacotes, geração de relatórios e execução de jobs.
O módulo não é uma biblioteca geral de grafos. Ele não calcula caminhos mínimos, centralidade, componentes ou fluxos. Seu foco é representar dependências e produzir uma ordem válida, incluindo uma API incremental que permite executar vários nós prontos em paralelo.
Modelo de dependências
TopologicalSorter recebe um mapping em que cada chave é um nó e o valor é um iterable com os predecessores desse nó.
from graphlib import TopologicalSorter
grafo = {
"testar": {"instalar"},
"empacotar": {"testar"},
"publicar": {"empacotar"},
"instalar": set(),
}
ordem = tuple(TopologicalSorter(grafo).static_order())
print(ordem)
A relação é “esta tarefa depende destas outras”. É comum inverter acidentalmente o sentido e obter uma sequência oposta ao esperado.
Nós hashable
Os nós precisam ser hashable, porque são usados em dicionários e sets. Strings, números, enums e tuplas imutáveis são opções naturais.
Objetos mutáveis com hash instável não devem ser utilizados. Se a tarefa possui muitos metadados, use um identificador imutável como nó e mantenha os detalhes em outro mapping.
Dependências sem chave explícita
Um predecessor citado nos valores não precisa aparecer originalmente como chave. O sorter o adiciona como nó sem predecessores conhecidos.
grafo = {
"deploy": {"build"},
}
Nesse exemplo, build também pertence ao grafo. Apesar da conveniência, declarar todos os nós explicitamente melhora validação e documentação.
static_order
static_order() é a forma mais simples de obter uma ordem completa.
sorter = TopologicalSorter(grafo)
for tarefa in sorter.static_order():
executar(tarefa)
O resultado é um iterator. Consuma-o uma única vez ou transforme em tupla quando precisar reutilizar a sequência.
Mais de uma ordem pode ser válida
Quando duas tarefas não dependem uma da outra, qualquer ordem relativa entre elas pode ser válida. Não escreva testes que exijam uma sequência total específica sem necessidade.
Valide as relações: para cada aresta, o predecessor deve aparecer antes do dependente. Quando a apresentação precisa ser determinística, ordene os nós prontos por uma chave estável fora do sorter.
Detecção de ciclos
Uma ordenação topológica só existe em grafos acíclicos. Quando há ciclo, o módulo lança CycleError.
from graphlib import CycleError, TopologicalSorter
try:
ordem = tuple(TopologicalSorter({
"a": {"b"},
"b": {"a"},
}).static_order())
except CycleError as erro:
print("ciclo detectado", erro)
Um ciclo significa que nenhuma das tarefas envolvidas pode começar sem outra do mesmo grupo.
Diagnóstico de ciclos
A exceção inclui informações que ajudam a identificar um ciclo, mas o formato não deve ser tratado como API de relatório completa. Para uma interface amigável, converta os nós em nomes e mostre uma cadeia clara.
Um grafo pode ter vários ciclos. Corrigir o primeiro pode revelar outros, portanto repita a validação após cada alteração.
Adicione nós incrementalmente
add(node, *predecessors) adiciona ou amplia dependências antes da preparação.
sorter = TopologicalSorter()
sorter.add("compilar", "gerar_codigo")
sorter.add("testar", "compilar")
sorter.add("publicar", "testar")
Chamadas repetidas para o mesmo nó unem os predecessores. Isso é útil quando plugins contribuem com dependências.
prepare
Para a API incremental, chame prepare() depois de montar o grafo. A preparação valida a estrutura e habilita get_ready() e done().
sorter.prepare()
Não adicione novas dependências depois que a execução foi preparada. Construa e valide toda a configuração primeiro.
get_ready
get_ready() devolve os nós cujos predecessores já foram marcados como concluídos.
prontas = sorter.get_ready()
for tarefa in prontas:
iniciar(tarefa)
Os nós retornados passam a estar “em andamento”. Eles não serão devolvidos novamente; o coordenador precisa chamar done() quando terminarem.
done
done(*nodes) informa que tarefas anteriormente prontas foram concluídas.
sorter.done("gerar_codigo")
novas = sorter.get_ready()
Não marque como concluído um nó que nunca foi devolvido por get_ready(). A classe valida transições inválidas e gera erros.
is_active
is_active() informa se ainda há trabalho que pode progredir: nós prontos não obtidos ou nós em andamento que ainda podem ser concluídos.
while sorter.is_active():
for tarefa in sorter.get_ready():
executar(tarefa)
sorter.done(tarefa)
Em execução paralela, tarefas terminadas chegam por uma fila de resultados.
Execução paralela
A API incremental permite enviar todos os nós prontos a workers, aguardar conclusões e liberar dependentes.
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
sorter.prepare()
ativos = {}
with ThreadPoolExecutor(max_workers=4) as executor:
while sorter.is_active():
for tarefa in sorter.get_ready():
ativos[executor.submit(executar, tarefa)] = tarefa
concluidos, _ = wait(ativos, return_when=FIRST_COMPLETED)
for futuro in concluidos:
tarefa = ativos.pop(futuro)
futuro.result()
sorter.done(tarefa)
O exemplo precisa de política adicional para falhas, cancelamento e shutdown.
Falhas de tarefas
Uma tarefa que falha não deve ser marcada como concluída automaticamente se seus dependentes exigem o resultado. O coordenador precisa decidir entre interromper o pipeline, marcar dependentes como bloqueados ou usar um resultado de fallback.
Registre o estado separadamente: concluída, falhou, cancelada e ignorada não são equivalentes.
Cancelamento
Ao cancelar uma execução, pare de enviar novos nós, solicite cancelamento cooperativo dos workers e aguarde um prazo. Não chame done() para tarefas incompletas apenas para esvaziar o sorter.
Se a aplicação quiser continuar com dependências opcionais, modele explicitamente essa regra fora do grafo obrigatório.
Dependências opcionais
TopologicalSorter representa precedência obrigatória. Dependência opcional, condição de feature flag ou branch de ambiente exige uma etapa anterior que construa o grafo efetivo.
Remova nós desativados e ajuste dependências antes de chamar prepare().
Validação de referências
Como predecessores não declarados são adicionados automaticamente, um erro de digitação pode criar uma tarefa fantasma.
declaradas = set(configuracao)
referenciadas = set().union(*configuracao.values())
desconhecidas = referenciadas - declaradas
Decida se nós implícitos são permitidos. Em pipelines configurados por usuários, geralmente é melhor rejeitar nomes desconhecidos.
Representação de tarefas
Separe a identidade da função executora.
tarefas = {
"extrair": lambda: extrair(dados),
"transformar": transformar,
"carregar": carregar,
}
Isso mantém o grafo serializável, facilita logs e permite validar nomes sem executar código.
Side effects e idempotência
Uma ordem válida não protege contra efeitos parciais. Se uma tarefa grava arquivo, banco ou API, defina idempotência, transação e rollback.
Ao retomar um pipeline, não presuma que uma tarefa com log de início terminou corretamente. Grave checkpoints após sucesso verificável.
Prioridades
O módulo não implementa prioridade entre nós simultaneamente prontos. O coordenador pode ordenar a tupla de get_ready() antes do envio.
for tarefa in sorted(sorter.get_ready(), key=prioridade):
iniciar(tarefa)
A prioridade não pode violar dependências; ela só escolhe entre opções já liberadas.
Limite de concorrência
Nem todo nó pronto deve iniciar imediatamente. Respeite o número de workers, conexões de banco, memória e rate limits.
Mantenha uma fila local de prontos ou envie apenas até a capacidade disponível. O artigo sobre concurrent.futures no Python explica backpressure e pools.
Recursos por categoria
Algumas tarefas usam CPU, outras banco ou rede. Um único limite global pode ser inadequado. Mantenha semáforos ou executors separados por tipo de recurso.
O grafo define precedência, enquanto o scheduler define capacidade.
Persistência do estado
TopologicalSorter não é um motor durável. Se o processo reiniciar, o estado em memória é perdido.
Para pipelines críticos, persista o grafo, tentativas, outputs, timestamps e status em armazenamento transacional. Reconstrua o sorter apenas para calcular quais tarefas podem avançar.
Grafos grandes
A ordenação possui custo proporcional ao número de nós e arestas em condições normais, mas memória e representação dos sets importam. Evite duplicar grandes estruturas sem necessidade.
Use IDs compactos e valide limites antes de aceitar configurações externas.
Entrada não confiável
Um usuário pode enviar um grafo enorme, com muitos predecessores ou nomes gigantes, causando consumo de memória e CPU.
Limite nós, arestas, tamanho de identificadores e tempo de preparação. Não associe nomes diretamente a imports ou comandos sem uma allowlist.
Visualização
O módulo não gera diagramas. Para depuração, exporte arestas para DOT, JSON ou uma tabela. Destaque ciclos, tarefas bloqueadas e nós sem dependentes.
Uma visualização é especialmente útil quando plugins adicionam relações automaticamente.
Testes
Teste grafo vazio, um nó, cadeia linear, branches independentes, convergência, múltiplas ordens válidas, ciclo, referência desconhecida, falha de tarefa, cancelamento e execução paralela.
Em testes de ordem, verifique precedência em vez de comparar uma lista total quando houver liberdade.
Erros comuns
Os erros mais frequentes são inverter o sentido das dependências, exigir uma única ordem, não chamar done(), marcar tarefa falha como concluída, adicionar nós depois de preparar, aceitar typo como nó implícito, iniciar trabalho ilimitado e tratar o sorter como sistema persistente de workflow.
Conclusão
graphlib.TopologicalSorter resolve de forma direta a ordenação de tarefas com dependências e oferece uma API incremental adequada a schedulers paralelos. Modele predecessores corretamente, valide ciclos e nomes, limite concorrência e mantenha o estado de execução separado do grafo.
Consulte a documentação oficial de graphlib e o guia de multiprocessing no Python quando as tarefas exigirem processos.
