graphlib.TopologicalSorter é uma classe da biblioteca padrão do Python criada para ordenar dependências em um grafo direcionado acíclico. Ela resolve um problema comum em sistemas reais: executar tarefas apenas depois que todos os seus pré-requisitos forem concluídos. Compilação de projetos, pipelines de dados, migrações de banco, geração de relatórios, processamento de arquivos e automações de infraestrutura são exemplos em que essa ordem importa.
Em vez de manter listas manuais e frágeis, você descreve quais nós dependem de quais outros. O TopologicalSorter calcula uma ordem válida e também oferece uma API incremental para distribuir tarefas entre workers paralelos sem violar dependências.
O que é ordenação topológica
Uma ordenação topológica é uma sequência de vértices na qual cada dependência aparece antes do item que depende dela. Considere um processo com as etapas “baixar dados”, “validar”, “transformar” e “publicar”. A validação depende do download; a transformação depende da validação; a publicação depende da transformação. A ordem resultante é direta. Em grafos maiores, várias tarefas podem ficar disponíveis ao mesmo tempo.
A técnica funciona apenas para grafos sem ciclos. Se A depende de B e B depende de A, nenhuma ordem válida existe. O Python detecta essa situação e levanta CycleError.
Primeiro exemplo com static_order
from graphlib import TopologicalSorter
grafo = {
"publicar": {"transformar"},
"transformar": {"validar"},
"validar": {"baixar"},
"baixar": set(),
}
ordem = list(TopologicalSorter(grafo).static_order())
print(ordem)
O dicionário usa cada tarefa como chave e o conjunto de predecessores como valor. static_order() prepara o grafo, verifica ciclos e produz uma ordem válida. Para scripts sequenciais, essa é a abordagem mais simples.
A ordem entre tarefas independentes não deve ser tratada como uma promessa de negócio. Se duas tarefas não dependem uma da outra, ambas podem aparecer em posições diferentes sem que o resultado esteja errado.
Construindo o grafo com add
Você também pode montar o grafo gradualmente:
from graphlib import TopologicalSorter
ts = TopologicalSorter()
ts.add("baixar")
ts.add("validar", "baixar")
ts.add("transformar", "validar")
ts.add("publicar", "transformar")
print(tuple(ts.static_order()))
O método add(no, *predecessores) acumula dependências. Chamar add novamente para o mesmo nó acrescenta novos predecessores. Isso é útil quando plugins, módulos ou arquivos de configuração contribuem para o mesmo pipeline.
Executando tarefas em paralelo
A API incremental foi projetada para concorrência. Primeiro chame prepare(). Depois, get_ready() retorna os nós cujas dependências já foram satisfeitas. Quando um worker termina uma tarefa, chame done(). Novos nós podem então ficar disponíveis.
from graphlib import TopologicalSorter
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
requisitos = {
"extrair_clientes": set(),
"extrair_pedidos": set(),
"unir": {"extrair_clientes", "extrair_pedidos"},
"relatorio": {"unir"},
}
def executar(nome):
print("executando", nome)
return nome
ts = TopologicalSorter(requisitos)
ts.prepare()
com ThreadPoolExecutor(max_workers=4) as pool:
futuros = {}
while ts.is_active() or futuros:
for tarefa in ts.get_ready():
futuros[pool.submit(executar, tarefa)] = tarefa
if not futuros:
break
concluidos, _ = wait(futuros, return_when=FIRST_COMPLETED)
for futuro in concluidos:
tarefa = futuros.pop(futuro)
futuro.result()
ts.done(tarefa)
Esse padrão permite paralelismo somente entre tarefas prontas. A chamada future.result() também propaga exceções do worker. Em produção, defina claramente o que fazer quando uma tarefa falhar: interromper tudo, tentar novamente, marcar dependentes como bloqueados ou aplicar compensação.
Prepare, get_ready e done
prepare() finaliza a preparação do grafo e detecta ciclos. Depois disso, você não deve adicionar novos nós. get_ready() pode retornar vários itens, e cada item retornado precisa ser posteriormente informado a done() quando tiver sido processado com sucesso.
Não marque uma tarefa como concluída antes de sua execução real. Fazer isso libera dependentes cedo demais e transforma um grafo correto em uma execução incorreta. Também não chame done() duas vezes para o mesmo nó.
is_active() indica se ainda existe trabalho possível ou em andamento. Ele ajuda a controlar o loop de um scheduler, especialmente quando tarefas são enviadas para filas externas.
Detectando ciclos
from graphlib import TopologicalSorter, CycleError
grafo = {
"a": {"b"},
"b": {"c"},
"c": {"a"},
}
try:
list(TopologicalSorter(grafo).static_order())
except CycleError as erro:
print("ciclo detectado", erro.args)
Ciclos normalmente apontam para erro de modelagem ou configuração. A mensagem pode trazer nós envolvidos, mas não dependa de uma representação específica para lógica crítica. O melhor diagnóstico é registrar as dependências carregadas e apresentar ao usuário uma cadeia compreensível.
Normalizando configurações
Dados de YAML, JSON ou banco podem conter duplicatas, nós ausentes e strings vazias. Normalize antes de criar o sorter.
def normalizar(configuracao):
grafo = {}
for nome, dependencias in configuracao.items():
nome = nome.strip()
if not nome:
raise ValueError("tarefa sem nome")
grafo[nome] = {d.strip() for d in dependencias if d.strip()}
return grafo
O TopologicalSorter aceita predecessores que não aparecem como chaves e os inclui no grafo. Isso pode ser conveniente, mas também pode esconder erro de digitação. Em sistemas rigorosos, valide se todos os nomes pertencem ao catálogo conhecido.
Exemplo de pipeline de arquivos
from graphlib import TopologicalSorter
pipeline = {
"compactar": {"gerar_csv", "gerar_pdf"},
"enviar": {"compactar"},
"gerar_csv": {"consultar"},
"gerar_pdf": {"consultar", "carregar_template"},
"consultar": set(),
"carregar_template": set(),
}
for etapa in TopologicalSorter(pipeline).static_order():
print(etapa)
“consultar” e “carregar_template” podem ser executadas primeiro, possivelmente em paralelo. “gerar_pdf” aguarda ambas, enquanto “gerar_csv” depende apenas da consulta. “compactar” espera os dois artefatos, e “enviar” encerra o fluxo.
Resultados, não apenas nomes
O sorter organiza nós, mas não armazena automaticamente resultados. Mantenha um dicionário compartilhado ou um armazenamento externo:
resultados = {}
def executar(tarefa):
if tarefa == "consultar":
resultados[tarefa] = [1, 2, 3]
elif tarefa == "gerar_csv":
resultados[tarefa] = criar_csv(resultados["consultar"])
Com threads, proteja estruturas mutáveis quando houver escrita concorrente. Com processos ou workers distribuídos, use banco, cache, fila ou object storage. Sempre associe o resultado a um identificador de execução para evitar mistura entre pipelines.
Idempotência e retomada
Um bom orquestrador precisa lidar com reinícios. Faça as tarefas idempotentes quando possível: executar novamente deve produzir o mesmo estado final. Registre status como pendente, executando, concluído e falhou. Ao retomar, reconstrua o grafo e marque como concluídas apenas tarefas cujo resultado foi realmente verificado.
O TopologicalSorter não persiste estado e não substitui Airflow, Prefect, Celery ou sistemas semelhantes. Ele é um componente leve para aplicações que precisam controlar dependências localmente ou implementar uma camada própria de agendamento.
Cuidados com memória e escala
O grafo permanece em memória. Para milhares ou dezenas de milhares de nós, isso costuma ser aceitável, mas meça. Grafos enormes e dinâmicos podem exigir particionamento ou um mecanismo especializado. Além disso, a ordenação não reduz o custo real das tarefas; ela apenas determina quando cada uma pode começar.
Evite usar objetos mutáveis como nós. Strings, inteiros, enums e tuplas imutáveis são escolhas previsíveis. Um nó precisa ser hashable porque é usado em dicionários e conjuntos.
Boas práticas
Use nomes estáveis, valide dependências desconhecidas, detecte ciclos antes de iniciar trabalho caro, limite o número de workers, propague falhas, registre duração e resultados e não confunda “pronto” com “concluído”. Para pipelines importantes, salve eventos de execução e inclua identificadores de correlação nos logs.
Veja também os conteúdos da Academify sobre dicionários em Python, conjuntos em Python, threads em Python e concurrent.futures. Como referências externas, consulte a documentação oficial de graphlib e a documentação de concurrent.futures.
Conclusão
graphlib.TopologicalSorter transforma uma rede de dependências em uma ordem executável e oferece uma base segura para paralelismo controlado. Para fluxos pequenos e médios, ele elimina listas manuais, detecta ciclos e deixa explícito o contrato entre tarefas. Quando combinado com validação, persistência, idempotência e tratamento de falhas, torna-se uma peça valiosa para construir pipelines confiáveis usando apenas a biblioteca padrão do Python.
