El módulo graphlib ofrece la clase TopologicalSorter para ordenar tareas que tienen dependencias. Un orden topológico organiza los nodos de un grafo dirigido acíclico de manera que cada requisito aparezca antes del elemento que depende de él. Este patrón aparece en sistemas de build, pipelines de datos, migraciones, instalación de paquetes, generación de informes y ejecución de jobs.
El módulo no es una biblioteca general de grafos. No calcula caminos mínimos, centralidad, componentes o flujos. Su objetivo es representar dependencias y producir un orden válido, además de una API incremental que permite ejecutar en paralelo varios nodos listos.
Modelo de dependencias
TopologicalSorter recibe un mapping donde cada clave es un nodo y el valor es un iterable con sus predecesores.
from graphlib import TopologicalSorter
grafo = {
"probar": {"instalar"},
"empaquetar": {"probar"},
"publicar": {"empaquetar"},
"instalar": set(),
}
orden = tuple(TopologicalSorter(grafo).static_order())
print(orden)
La relación significa “esta tarea depende de estos predecesores”. Invertir el sentido es un error frecuente.
Nodos hashable
Los nodos deben ser hashable porque se almacenan en diccionarios y sets. Strings, números, enums y tuplas inmutables son opciones naturales.
No uses objetos mutables con hash inestable. Mantén un identificador compacto e inmutable en el grafo y guarda los metadatos en otro mapping.
Predecesores implícitos
Un predecesor citado en los valores no necesita aparecer como clave explícita. El sorter lo añade como nodo sin predecesores conocidos.
grafo = {
"deploy": {"build"},
}
Aquí build también forma parte del grafo. La comodidad es útil, pero declarar todos los nodos mejora validación y documentación.
static_order
static_order() es la forma más simple de producir un orden completo válido.
sorter = TopologicalSorter(grafo)
for tarea in sorter.static_order():
ejecutar(tarea)
El resultado es un iterator. Consúmelo una vez o conviértelo en tupla si necesitas reutilizar la secuencia.
Puede haber varios órdenes válidos
Cuando dos tareas son independientes, cualquiera de los dos órdenes relativos satisface el grafo. Los tests no deberían exigir una secuencia total exacta salvo que la aplicación añada otra regla determinística.
Valida la precedencia: cada predecesor debe aparecer antes que su dependiente. Ordena los nodos listos por una clave estable cuando necesites presentación reproducible.
Detección de ciclos
Un orden topológico solo existe en un grafo acíclico. Un ciclo genera CycleError.
from graphlib import CycleError, TopologicalSorter
try:
orden = tuple(TopologicalSorter({
"a": {"b"},
"b": {"a"},
}).static_order())
except CycleError as error:
print("ciclo detectado", error)
Un ciclo significa que ninguna de sus tareas puede comenzar porque cada una espera a otra del mismo grupo.
Informe de ciclos
La excepción contiene información útil para localizar un ciclo, pero no debe tratarse como una API completa de presentación. Convierte IDs en nombres legibles y muestra una cadena clara.
Un grafo puede tener varios ciclos. Corregir el primero puede revelar otro, así que vuelve a validar.
Añade nodos incrementalmente
add(node, *predecessors) añade o amplía dependencias antes de preparar.
sorter = TopologicalSorter()
sorter.add("compilar", "generar_codigo")
sorter.add("probar", "compilar")
sorter.add("publicar", "probar")
Llamadas repetidas para el mismo nodo unen los predecesores. Es útil cuando plugins contribuyen con relaciones.
prepare
La API incremental comienza con prepare(). La llamada valida el grafo y habilita get_ready() y done().
sorter.prepare()
No añadas nuevas dependencias después de preparar. Construye y valida toda la configuración primero.
get_ready
get_ready() devuelve los nodos cuyos predecesores ya fueron marcados como terminados.
listas = sorter.get_ready()
for tarea in listas:
iniciar(tarea)
Los nodos devueltos pasan a estar “en progreso”. No volverán a aparecer; el coordinador debe llamar done() al terminar.
done
done(*nodes) marca como completos nodos previamente listos.
sorter.done("generar_codigo")
nuevas = sorter.get_ready()
No marques un nodo que nunca fue devuelto por get_ready(). Las transiciones inválidas generan errores.
is_active
is_active() indica si todavía puede haber progreso: nodos listos no obtenidos o nodos en ejecución que pueden concluir.
while sorter.is_active():
for tarea in sorter.get_ready():
ejecutar(tarea)
sorter.done(tarea)
En ejecución paralela, las tareas concluidas suelen llegar por una queue o un conjunto de futures.
Ejecución paralela
La API incremental permite enviar todos los nodos listos a workers, esperar resultados y liberar dependientes.
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
sorter.prepare()
activas = {}
with ThreadPoolExecutor(max_workers=4) as executor:
while sorter.is_active():
for tarea in sorter.get_ready():
activas[executor.submit(ejecutar, tarea)] = tarea
concluidas, _ = wait(activas, return_when=FIRST_COMPLETED)
for futuro in concluidas:
tarea = activas.pop(futuro)
futuro.result()
sorter.done(tarea)
Una versión de producción también necesita políticas de error, cancelación y shutdown.
Fallos de tareas
Una tarea fallida no debe marcarse automáticamente como completa cuando sus dependientes requieren un resultado válido. El coordinador decide si detiene el pipeline, bloquea descendientes o usa un fallback documentado.
Mantén estados distintos: éxito, fallo, cancelación, skip y bloqueo no son equivalentes.
Cancelación
Al cancelar, deja de enviar nuevos nodos, solicita cancelación cooperativa y espera un plazo. No llames done() sobre tareas incompletas para vaciar el sorter.
Las rutas opcionales deben modelarse explícitamente fuera del grafo obligatorio.
Dependencias opcionales
TopologicalSorter representa precedencia obligatoria. Feature flags, condiciones de entorno y plugins opcionales requieren una etapa que construya el grafo efectivo antes de prepare().
Elimina nodos desactivados y ajusta relaciones antes de ejecutar.
Valida referencias
Como los predecesores ausentes se convierten en nodos implícitos, un typo puede crear una tarea fantasma.
declaradas = set(configuracion)
referenciadas = set().union(*configuracion.values())
desconocidas = referenciadas - declaradas
Decide si permites nodos implícitos. Los pipelines configurados por usuarios suelen beneficiarse de rechazar nombres desconocidos.
Separa identidad y ejecución
Guarda IDs de tareas en el grafo y funciones en otro mapping.
tareas = {
"extraer": lambda: extraer(datos),
"transformar": transformar,
"cargar": cargar,
}
Esto mantiene el grafo serializable, mejora logs y evita ejecutar código durante validación.
Side effects e idempotencia
Un orden válido no protege frente a escrituras parciales. Las tareas que modifican archivos, bases o APIs necesitan idempotencia, transacciones y rollback.
Al reanudar un pipeline, no supongas que un log de inicio significa éxito. Persiste checkpoints después de un resultado verificable.
Prioridades
El módulo no prioriza nodos listos simultáneamente. El coordinador puede ordenar la tupla devuelta por get_ready().
for tarea in sorted(sorter.get_ready(), key=prioridad):
iniciar(tarea)
La prioridad elige entre nodos liberados, pero no puede violar dependencias.
Límites de concurrencia
No todas las tareas listas deben comenzar inmediatamente. Respeta workers, pool de base, memoria, ancho de banda y rate limits.
Mantén una queue local o envía solo hasta la capacidad disponible. Consulta concurrent.futures en Python para backpressure.
Clases de recursos
Algunas tareas consumen CPU, otras base o red. Un único límite global puede ser inadecuado. Usa executors o semáforos separados por recurso.
El grafo describe precedencia; el scheduler describe capacidad.
Estado persistente
TopologicalSorter no es un motor durable. Un reinicio pierde el progreso en memoria.
Los pipelines críticos deben persistir grafo, intentos, outputs, timestamps y estados. Reconstruye el sorter solo para calcular qué tareas pueden avanzar.
Grafos grandes
La ordenación suele escalar con nodos y aristas, pero la representación y memoria importan. Evita duplicar estructuras grandes.
Usa IDs compactos y límites antes de aceptar configuración externa.
Entrada no confiable
Un usuario puede enviar millones de aristas, nombres enormes o fan-in extremo y consumir CPU y memoria.
Limita nodos, aristas, longitud de IDs y tiempo de preparación. No conviertas nombres directamente en imports o comandos sin allowlist.
Visualización
El módulo no genera diagramas. Exporta aristas a DOT, JSON o tabla. Destaca ciclos, tareas bloqueadas y nodos finales.
La visualización es especialmente útil cuando plugins añaden dependencias automáticamente.
Pruebas
Prueba grafo vacío, un nodo, cadena lineal, branches independientes, convergencia, varios órdenes válidos, ciclos, referencias desconocidas, fallos, cancelación y finalización paralela.
Cuando el orden no sea único, verifica restricciones de precedencia en vez de una lista exacta.
Errores comunes
Los fallos frecuentes son invertir dependencias, exigir un único orden, olvidar done(), marcar fallos como completos, añadir nodos tras preparar, aceptar typos como tareas, lanzar trabajo ilimitado y tratar el sorter como sistema persistente.
Conclusión
graphlib.TopologicalSorter resuelve la ordenación de dependencias y ofrece una interfaz incremental adecuada para schedulers paralelos. Modela correctamente los predecesores, valida ciclos y nombres, limita concurrencia y separa el estado de ejecución del grafo.
Consulta la documentación oficial de graphlib y multiprocessing en Python cuando las tareas necesiten procesos.
