graphlib en Python: Orden Topológico de Tareas

graphlib.TopologicalSorter resuelve un problema frecuente en builds, pipelines y migraciones: ordenar tareas respetando dependencias. La entrada relaciona cada nodo con sus predecesores, no con sus sucesores.

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])

Usa static_order() cuando basta una secuencia. Si existe un ciclo se lanza CycleError. No trates el orden exacto entre nodos independientes como contrato, porque puede haber varias secuencias válidas.

Buenas prácticas

Para ejecución paralela, combina prepare(), get_ready() y done() para liberar tareas cuando terminan sus predecesores. Usa nodos hashable y valida configuraciones externas antes de construir el grafo. Un ciclo suele indicar una configuración inválida y debe detener la ejecución.

Para trabajo por prioridad, consulta heapq en Python. Si las tareas son corutinas relacionadas, asyncio.TaskGroup controla fallos y cancelaciones.

La documentación oficial de graphlib, consultada el 22 de julio de 2026, detalla la API y sus garantías.

Modelar dependencias

Cada clave del mapeo es un nodo y sus valores son predecesores que deben terminar primero. Para indicar que compilar depende de generar, escribe {"compilar": {"generar"}}. Es fácil invertir esta dirección porque otros formatos enumeran sucesores. Un predecesor ausente como clave también se convierte en nodo.

Usa identificadores hashable y estables, como cadenas o enums, y guarda comandos en otro mapeo. Así puedes validar y mostrar el grafo sin efectos.

from graphlib import TopologicalSorter

grafo = {
    "obtener": set(),
    "validar": {"obtener"},
    "transformar": {"validar"},
    "informe": {"validar"},
    "publicar": {"transformar", "informe"},
}
orden = list(TopologicalSorter(grafo).static_order())
posicion = {nodo: i for i, nodo in enumerate(orden)}
assert posicion["validar"] < posicion["transformar"]
assert posicion["informe"] < posicion["publicar"]

Prueba relaciones obligatorias, no una secuencia completa. Los nodos independientes pueden intercambiar posiciones.

Ciclos y fallos

Un ciclo significa que no existe orden completo válido. Puede ser una autorreferencia o una cadena larga que regresa al inicio. CycleError aporta detalles en args, pero la aplicación debe convertirlos en un mensaje claro.

Tras detectar un ciclo, algunos nodos ajenos pueden seguir disponibles. Esto ayuda al diagnóstico, pero builds y migraciones normalmente deben fallar antes de producir efectos. Ejecutar solo una parte deja estados inesperados.

Ejecución paralela

Usa prepare(), get_ready() y done() para aprovechar trabajo independiente. Envía nodos listos y comunica su finalización únicamente tras éxito.

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

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

Llama a done() una sola vez y únicamente después del éxito. Si result() falla, los dependientes permanecen bloqueados. El código real debe cancelar pendientes y registrar tareas completadas.

Ciclo de vida y validación

Añade dependencias antes de ordenar. Después de prepare(), trata el grafo como cerrado y crea otro sorter si cambia la configuración. Normaliza entradas externas y no modifiques conjuntos durante su uso.

Las aristas duplicadas no importan, pero una ausente puede liberar trabajo antes de tiempo. Valida referencias, nombres vacíos, etapas deshabilitadas y dependencias condicionales. Limita nodos y aristas en entradas no confiables.

Determinismo

Un DAG admite varios órdenes. No conviertas el orden observado entre nodos independientes en contrato. Para logs reproducibles, normaliza la entrada y ordena solo cada lote listo. Ordenar alfabéticamente el resultado final puede romper dependencias. La finalización real seguirá variando en paralelo.

Alcance y complejidad

La ordenación recorre nodos y aristas, así que el coste escala con el tamaño. Ejecutar tareas suele ser más caro. graphlib no es un scheduler completo: no ofrece persistencia, reintentos, timeout, prioridad, bloqueo de recursos ni coordinación distribuida. Solo identifica trabajo ejecutable.

Sirve para ETL, builds, inicialización, migraciones y prerrequisitos. Prioridad y preferencia no son necesariamente dependencias y requieren políticas separadas.

Estrategia de pruebas

Cubre grafo vacío, nodo aislado, cadena, bifurcación, convergencia y ciclo. Comprueba que cada predecesor aparece antes de su dependiente y cada nodo una vez. En modo incremental, verifica que done() libera exactamente los sucesores esperados. Simula un fallo sin done() y confirma que los dependientes no comienzan.

Registra duración, estado y error por nodo. Si las tareas producen efectos, hazlas idempotentes o define compensación, pues una puede fallar tras finalizar otras.

Lista práctica

Confirma dirección, normaliza identificadores y rechaza ciclos antes de efectos irreversibles. No dependas del orden entre nodos independientes. Usa done() solo tras éxito. Define cancelación y estado parcial. Para flujos durables o distribuidos, usa TopologicalSorter para planificar y añade infraestructura de ejecución adecuada.