graphlib.TopologicalSorter es una clase de la biblioteca estándar de Python diseñada para ordenar dependencias en un grafo dirigido acíclico. Resuelve un problema habitual en sistemas reales: una tarea solo puede ejecutarse después de que todos sus requisitos previos hayan finalizado. Compilación de proyectos, pipelines de datos, migraciones de bases de datos, generación de informes, despliegues y procesamiento de archivos son ejemplos claros.
En lugar de mantener una secuencia manual y frágil, describes qué nodos dependen de cuáles predecesores. TopologicalSorter calcula un orden válido y también ofrece una API incremental para enviar tareas listas a workers paralelos sin romper las dependencias.
Qué es un orden topológico
Un orden topológico es una secuencia en la que cada dependencia aparece antes del elemento que depende de ella. Imagina cuatro pasos: descargar datos, validarlos, transformarlos y publicarlos. La validación depende de la descarga; la transformación depende de la validación; y la publicación depende de la transformación. En un grafo grande, varias tareas independientes pueden quedar disponibles al mismo tiempo.
La ordenación topológica solo funciona con grafos sin ciclos. Si A depende de B y B depende de A, no existe un orden válido. Python detecta esta situación y lanza CycleError.
Primer ejemplo con static_order
from graphlib import TopologicalSorter
grafo = {
"publicar": {"transformar"},
"transformar": {"validar"},
"validar": {"descargar"},
"descargar": set(),
}
orden = list(TopologicalSorter(grafo).static_order())
print(orden)
El diccionario relaciona cada tarea con su conjunto de predecesores. static_order() prepara el grafo, comprueba ciclos y produce una secuencia válida. Para scripts secuenciales, esta es la interfaz más sencilla.
No debes considerar el orden relativo de tareas independientes como una garantía de negocio. Si dos nodos no dependen entre sí, cualquiera puede aparecer primero y el resultado seguirá siendo correcto.
Construir el grafo con add
from graphlib import TopologicalSorter
ts = TopologicalSorter()
ts.add("descargar")
ts.add("validar", "descargar")
ts.add("transformar", "validar")
ts.add("publicar", "transformar")
print(tuple(ts.static_order()))
add(nodo, *predecesores) acumula dependencias. Si llamas de nuevo a add para el mismo nodo, se agregan nuevos predecesores. Esto resulta útil cuando plugins, módulos o archivos de configuración aportan partes diferentes del mismo flujo.
Ejecutar tareas en paralelo
La API incremental está pensada para concurrencia. Primero llama a prepare(). Después, get_ready() devuelve todos los nodos cuyas dependencias ya están satisfechas. Cuando un worker termina una tarea, llama a done(). Entonces pueden liberarse nuevos nodos.
from graphlib import TopologicalSorter
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
requisitos = {
"extraer_clientes": set(),
"extraer_pedidos": set(),
"unir": {"extraer_clientes", "extraer_pedidos"},
"informe": {"unir"},
}
def ejecutar(nombre):
print("ejecutando", nombre)
return nombre
ts = TopologicalSorter(requisitos)
ts.prepare()
with ThreadPoolExecutor(max_workers=4) as pool:
futuros = {}
while ts.is_active() or futuros:
for tarea in ts.get_ready():
futuros[pool.submit(ejecutar, tarea)] = tarea
if not futuros:
break
terminados, _ = wait(futuros, return_when=FIRST_COMPLETED)
for futuro in terminados:
tarea = futuros.pop(futuro)
futuro.result()
ts.done(tarea)
Este patrón permite paralelismo solo entre tareas listas. future.result() propaga excepciones del worker. En producción debes definir la política de error: detener todo, reintentar, bloquear descendientes o ejecutar una compensación.
prepare, get_ready, done e is_active
prepare() finaliza la preparación y detecta ciclos. Después de iniciar la ejecución, el grafo debe considerarse fijo. get_ready() puede devolver varios nodos, y cada uno debe notificarse posteriormente mediante done() cuando termine correctamente.
Nunca marques una tarea como completada antes de que su trabajo real haya finalizado. Eso libera dependientes demasiado pronto. Tampoco llames dos veces a done() para el mismo nodo.
is_active() indica si todavía existe trabajo posible o pendiente. Es útil para controlar el bucle de un scheduler, especialmente cuando las tareas se envían a una cola externa.
Detectar ciclos
from graphlib import TopologicalSorter, CycleError
grafo = {
"a": {"b"},
"b": {"c"},
"c": {"a"},
}
try:
list(TopologicalSorter(grafo).static_order())
except CycleError as error:
print("ciclo detectado", error.args)
Los ciclos normalmente indican un error de modelado o configuración. La excepción puede incluir nodos relacionados, pero no conviene depender de una representación exacta. Un diagnóstico sólido registra las dependencias cargadas y muestra una cadena comprensible.
Normalizar configuraciones
Los datos procedentes de YAML, JSON o una base de datos pueden contener duplicados, nombres vacíos o dependencias desconocidas. Normaliza antes de crear el sorter.
def normalizar(configuracion):
grafo = {}
for nombre, dependencias in configuracion.items():
nombre = nombre.strip()
if not nombre:
raise ValueError("tarea sin nombre")
grafo[nombre] = {d.strip() for d in dependencias if d.strip()}
return grafo
TopologicalSorter acepta predecesores que no aparecen como claves y los incluye automáticamente como nodos. Esto puede ser útil, pero también puede ocultar un error tipográfico. En sistemas estrictos, valida cada dependencia contra un catálogo conocido.
Ejemplo de pipeline de archivos
from graphlib import TopologicalSorter
pipeline = {
"comprimir": {"crear_csv", "crear_pdf"},
"enviar": {"comprimir"},
"crear_csv": {"consultar"},
"crear_pdf": {"consultar", "cargar_plantilla"},
"consultar": set(),
"cargar_plantilla": set(),
}
for etapa in TopologicalSorter(pipeline).static_order():
print(etapa)
consultar y cargar_plantilla pueden ejecutarse primero, posiblemente en paralelo. crear_pdf espera ambas; crear_csv solo depende de la consulta. comprimir espera los dos artefactos y enviar cierra el flujo.
Gestionar resultados
El sorter organiza nodos, pero no almacena resultados. Usa un diccionario o un almacenamiento externo:
resultados = {}
def ejecutar(tarea):
if tarea == "consultar":
resultados[tarea] = [1, 2, 3]
elif tarea == "crear_csv":
resultados[tarea] = crear_csv(resultados["consultar"])
Con threads, protege estructuras mutables si hay escrituras concurrentes. Con procesos o workers distribuidos, usa una base de datos, caché, broker de mensajes u object storage. Relaciona los resultados con un identificador de ejecución para evitar mezclas entre pipelines.
Idempotencia y recuperación
Un orquestador fiable debe soportar reinicios. Haz que las tareas sean idempotentes siempre que sea posible: ejecutarlas de nuevo debe llevar al mismo estado final. Registra estados como pendiente, ejecutando, completado y fallido. Al recuperar, reconstruye el grafo y marca como completas solo las tareas cuyo resultado haya sido verificado.
TopologicalSorter no persiste estado y no sustituye Airflow, Prefect, Celery ni un motor especializado. Es un componente ligero para aplicaciones que necesitan controlar dependencias localmente o implementar su propio scheduler.
Reintentos y propagación de fallos
Los reintentos pertenecen a la capa de ejecución, no al grafo. Limita intentos, usa backoff exponencial para errores transitorios y no repitas validaciones que fallan de forma permanente. Los descendientes de una tarea fallida deben permanecer bloqueados salvo que la política indique lo contrario.
Guarda la excepción original, el número de intento, la hora de inicio y la hora de fin. Estos datos facilitan el diagnóstico y permiten distinguir una dependencia lenta de un problema real del scheduler.
Memoria y escala
El grafo permanece en memoria. Miles o decenas de miles de nodos pueden ser razonables, pero debes medir. Grafos enormes y dinámicos pueden requerir particionado o un sistema especializado. La ordenación tampoco reduce el coste de las tareas; solo determina cuándo pueden empezar.
Prefiere identificadores inmutables y hashables como strings, enteros, enums o tuplas. Los nodos se usan en diccionarios y conjuntos, por lo que los valores mutables no son adecuados.
Pruebas correctas
Prueba ramas independientes, un requisito compartido, un nodo con varios predecesores, una dependencia desconocida y al menos un ciclo intencional. En ejecución concurrente, verifica que ningún descendiente empiece antes de que todos sus predecesores hayan terminado.
Una prueba robusta valida relaciones, no un orden total exacto. Para cada arista dependencia -> tarea, comprueba que el índice de la dependencia sea menor que el de la tarea. Así la prueba sigue siendo válida aunque cambie la posición de nodos independientes.
Buenas prácticas
Usa nombres estables, valida dependencias desconocidas, detecta ciclos antes de iniciar trabajo costoso, limita workers, propaga fallos, registra duraciones y resultados y no confundas “lista” con “completada”. Los pipelines importantes deben persistir eventos e incluir identificadores de correlación en los logs.
También puedes consultar los artículos de Academify sobre diccionarios en Python, conjuntos en Python, threading en Python y concurrent.futures. Como fuentes externas, revisa la documentación oficial de graphlib y la documentación de concurrent.futures.
Conclusión
graphlib.TopologicalSorter convierte una red de dependencias en un orden ejecutable y ofrece una base segura para paralelismo controlado. En flujos pequeños y medianos elimina secuencias manuales, detecta ciclos y hace explícito el contrato entre tareas. Combinado con validación, persistencia, idempotencia y tratamiento de errores, es una pieza práctica para construir pipelines fiables con la biblioteca estándar de Python.
