Em muitos projetos, algumas tarefas só podem começar depois que outras terminam. Um pacote depende de bibliotecas, uma migração depende da anterior, uma etapa de ETL precisa aguardar a extração e um curso pode exigir disciplinas prévias. Esse tipo de relação forma um grafo direcionado de dependências. O módulo graphlib no Python, disponível na biblioteca padrão, oferece a classe TopologicalSorter para organizar esses itens em uma ordem válida e detectar ciclos que tornam a execução impossível.
Neste guia, você aprenderá a modelar dependências, gerar uma ordenação topológica, processar tarefas em paralelo, tratar ciclos e testar o resultado. O assunto se conecta a estruturas e técnicas já usadas em contextvars no Python, filas de prioridade com heapq, cached_property e singledispatch.
O que é ordenação topológica?
Uma ordenação topológica é uma sequência linear de vértices de um grafo direcionado na qual cada dependência aparece antes do item que depende dela. Se a tarefa publicar depende de testar, então testar deve surgir primeiro. Essa ordem só existe quando o grafo não possui ciclos direcionados.
Um ciclo ocorre quando A depende de B, B depende de C e C depende novamente de A. Nesse cenário, nenhuma tarefa pode ser a primeira, porque todas aguardam outra etapa do mesmo ciclo. O TopologicalSorter detecta essa condição e lança CycleError.
Como representar dependências
A classe recebe um dicionário em que cada chave é um nó e o valor é um conjunto ou iterável com seus predecessores. Em outras palavras, você descreve de quais itens cada tarefa depende:
from graphlib import TopologicalSorter
dependencias = {
"publicar": {"testar"},
"testar": {"compilar"},
"compilar": {"instalar_dependencias"},
"instalar_dependencias": set(),
}
ordem = list(TopologicalSorter(dependencias).static_order())
print(ordem)O resultado respeita todas as relações. A ordem exata pode variar quando existem tarefas independentes, pois mais de uma sequência pode ser válida.
Usando static_order
Para fluxos simples e sequenciais, static_order() é a API mais direta. Ela prepara o grafo, verifica ciclos e devolve um iterador com os nós em ordem topológica:
from graphlib import TopologicalSorter
etapas = {
"enviar_email": {"gerar_relatorio"},
"gerar_relatorio": {"consultar_banco", "carregar_config"},
"consultar_banco": {"carregar_config"},
"carregar_config": set(),
}
for etapa in TopologicalSorter(etapas).static_order():
print(f"Executando: {etapa}")Essa abordagem é adequada quando cada etapa deve terminar antes da próxima, quando o custo é baixo ou quando você apenas precisa validar a estrutura de dependências.
Adicionando nós gradualmente
Também é possível criar o objeto vazio e chamar add(). O primeiro argumento é o nó; os seguintes são predecessores:
from graphlib import TopologicalSorter
ts = TopologicalSorter()
ts.add("build", "lint", "test")
ts.add("deploy", "build")
ts.add("lint")
ts.add("test", "install")
ts.add("install")
print(list(ts.static_order()))Chamadas repetidas para o mesmo nó acumulam dependências. Isso é útil quando diferentes partes da aplicação registram etapas de um pipeline. No entanto, depois que o grafo é preparado, ele não deve mais ser modificado.
Processamento paralelo
A maior vantagem do TopologicalSorter aparece quando várias tarefas independentes podem executar ao mesmo tempo. Para isso, use prepare(), get_ready(), done() e is_active().
from graphlib import TopologicalSorter
etapas = {
"deploy": {"testes", "build_frontend"},
"testes": {"instalar"},
"build_frontend": {"instalar"},
"instalar": set(),
}
ts = TopologicalSorter(etapas)
ts.prepare()
while ts.is_active():
prontas = ts.get_ready()
for tarefa in prontas:
print("Executando", tarefa)
# execute a tarefa e aguarde a conclusão
ts.done(tarefa)No exemplo, depois de instalar, as tarefas testes e build_frontend ficam prontas simultaneamente. Em um sistema real, elas podem ser enviadas para threads, processos ou trabalhadores distribuídos. O método done() informa que uma tarefa terminou e libera seus sucessores.
Exemplo com ThreadPoolExecutor
Quando as etapas envolvem operações de entrada e saída, um pool de threads pode aproveitar o paralelismo disponível:
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
from graphlib import TopologicalSorter
import time
etapas = {
"publicar": {"testar", "documentar"},
"testar": {"compilar"},
"documentar": {"compilar"},
"compilar": {"baixar"},
"baixar": set(),
}
def executar(nome):
print("início", nome)
time.sleep(0.2)
print("fim", nome)
return nome
ts = TopologicalSorter(etapas)
ts.prepare()
com_futuro = {}
with ThreadPoolExecutor(max_workers=3) as executor:
while ts.is_active():
for tarefa in ts.get_ready():
futuro = executor.submit(executar, tarefa)
com_futuro[futuro] = tarefa
concluidos, _ = wait(com_futuro, return_when=FIRST_COMPLETED)
for futuro in concluidos:
tarefa = com_futuro.pop(futuro)
futuro.result()
ts.done(tarefa)Em produção, trate exceções antes de chamar done(). Uma tarefa com erro não deve liberar automaticamente etapas que dependem dela. Você pode cancelar o fluxo, registrar a falha ou aplicar uma política de repetição.
Detectando ciclos
O método prepare() e o iterador de static_order() detectam ciclos. A exceção CycleError inclui informações sobre um ciclo encontrado:
from graphlib import CycleError, TopologicalSorter
grafo = {
"a": {"c"},
"b": {"a"},
"c": {"b"},
}
try:
ordem = list(TopologicalSorter(grafo).static_order())
except CycleError as erro:
print("Dependência circular:", erro.args)Mesmo quando existe um ciclo, o objeto pode liberar nós que não participam dele antes de ficar bloqueado. Para ferramentas de diagnóstico, isso permite processar partes válidas e depois informar o conjunto problemático.
Dependências ausentes
Se um predecessor é mencionado, mas não aparece como chave, ele é adicionado automaticamente como nó sem predecessores. Esse comportamento é conveniente, porém pode esconder erros de digitação. Se você escrever configuracao em um lugar e configuração em outro, o grafo terá dois nós distintos.
Uma validação prévia pode comparar todos os nomes com um catálogo autorizado:
catalogo = {"config", "extrair", "transformar", "carregar"}
dependencias = {
"transformar": {"extrair"},
"carregar": {"transformar", "config"},
}
referenciados = set(dependencias)
for predecessores in dependencias.values():
referenciados.update(predecessores)
desconhecidos = referenciados - catalogo
if desconhecidos:
raise ValueError(f"Tarefas desconhecidas: {desconhecidos}")Usando objetos como nós
Os nós precisam ser hashable, mas não precisam ser strings. Enums, tuplas e objetos imutáveis funcionam bem. Uma dataclass congelada ajuda a carregar metadados:
from dataclasses import dataclass
from graphlib import TopologicalSorter
@dataclass(frozen=True)
class Tarefa:
nome: str
equipe: str
extrair = Tarefa("extrair", "dados")
validar = Tarefa("validar", "qualidade")
publicar = Tarefa("publicar", "plataforma")
grafo = {
validar: {extrair},
publicar: {validar},
}
for tarefa in TopologicalSorter(grafo).static_order():
print(tarefa.nome, tarefa.equipe)Evite objetos mutáveis cujo hash possa mudar. Alterar um campo usado em igualdade ou hash depois que ele entrou no grafo torna o comportamento imprevisível.
Aplicações práticas
A ordenação topológica aparece em vários contextos:
- resolução da ordem de instalação de pacotes;
- pipelines de dados e machine learning;
- migrações de banco de dados;
- compilação de módulos;
- planejamento de tarefas com pré-requisitos;
- geração de planilhas e relatórios dependentes;
- ordenação de plugins e serviços;
- validação de currículos acadêmicos.
O módulo resolve a ordem, mas não oferece persistência, filas distribuídas, repetição automática nem controle de recursos. Para fluxos grandes, ele pode atuar como núcleo de planejamento combinado com um executor.
Como testar
Não teste apenas uma sequência fixa quando várias ordens são possíveis. Teste a propriedade: todo predecessor deve aparecer antes do nó dependente.
from graphlib import TopologicalSorter
def validar_ordem(grafo):
ordem = list(TopologicalSorter(grafo).static_order())
posicao = {no: indice for indice, no in enumerate(ordem)}
for no, predecessores in grafo.items():
for anterior in predecessores:
assert posicao[anterior] < posicao[no]
validar_ordem({
"deploy": {"teste", "build"},
"teste": {"install"},
"build": {"install"},
"install": set(),
})Crie ainda testes para grafo vazio, nó isolado, múltiplas dependências, ciclo simples, ciclo indireto e falha de uma tarefa executada em paralelo.
Erros comuns
- Inverter o formato e informar sucessores em vez de predecessores.
- Assumir que a ordem entre tarefas independentes será sempre a mesma.
- Esquecer de chamar
done()no processamento incremental. - Chamar
done()antes de a tarefa realmente terminar. - Modificar o grafo depois de
prepare(). - Ignorar exceções de trabalhadores e liberar dependentes incorretamente.
- Usar nomes inconsistentes e criar nós fantasmas.
- Esperar que a classe execute as tarefas por conta própria.
Boas práticas
- Use identificadores imutáveis e consistentes.
- Valide nós desconhecidos antes de preparar o grafo.
- Registre dependências em uma única direção: nó para predecessores.
- Separe planejamento, execução e persistência.
- Defina a política para falhas antes de paralelizar.
- Limite concorrência conforme CPU, memória e serviços externos.
- Registre início, fim, duração e resultado de cada tarefa.
- Teste propriedades do grafo, não apenas uma ordem específica.
Conclusão
O graphlib no Python transforma relações de dependência em uma ordem executável sem exigir bibliotecas externas. Com static_order(), você resolve fluxos sequenciais de forma simples. Com prepare(), get_ready(), done() e is_active(), pode liberar tarefas independentes para execução concorrente.
A parte mais importante é modelar corretamente os predecessores e tratar ciclos como erro de configuração. O TopologicalSorter não substitui um orquestrador completo, mas oferece uma base confiável para instaladores, pipelines, compiladores, migrações e qualquer sistema em que a ordem dependa de pré-requisitos.
Para detalhes de API e comportamento entre versões, consulte a documentação oficial de graphlib e a documentação de concurrent.futures.
