Muchos proyectos contienen tareas que no pueden comenzar hasta que otras hayan terminado. Un paquete depende de bibliotecas, un despliegue depende de las pruebas, una migración de base de datos depende de la anterior y una canalización de datos debe extraer información antes de transformarla. Estas relaciones forman un grafo dirigido de dependencias. El módulo estándar graphlib en Python incluye la clase TopologicalSorter, que permite obtener un orden válido, detectar trabajo que puede ejecutarse en paralelo e identificar dependencias circulares.
En esta guía aprenderás a modelar predecesores, generar una ordenación estática, coordinar trabajadores concurrentes, diagnosticar ciclos, validar nombres y probar el grafo. El tema complementa los artículos sobre contextvars en Python, colas de prioridad con heapq, cached_property y singledispatch.
Qué es una ordenación topológica
Una ordenación topológica es una secuencia lineal de los vértices de un grafo dirigido en la que cada requisito aparece antes que el elemento que depende de él. Si desplegar depende de probar, la prueba debe aparecer primero. Cuando existen tareas independientes, puede haber varias secuencias igualmente válidas.
Este orden solo existe en un grafo dirigido acíclico. Un ciclo aparece cuando A depende de B, B depende de C y C vuelve a depender de A. Ninguna tarea del ciclo puede comenzar porque todas esperan a otra. TopologicalSorter detecta la situación y produce CycleError.
Cómo representar las dependencias
El constructor recibe un diccionario donde cada clave es un nodo y su valor es un iterable con los predecesores. La dirección importa: los valores son requisitos, no tareas posteriores.
from graphlib import TopologicalSorter
dependencias = {
"desplegar": {"probar"},
"probar": {"compilar"},
"compilar": {"instalar"},
"instalar": set(),
}
orden = list(TopologicalSorter(dependencias).static_order())
print(orden)El resultado respeta todas las relaciones. La posición relativa de dos tareas independientes puede cambiar, por lo que no debes depender de un desempate accidental.
Usar static_order
Para un flujo secuencial, static_order() es la interfaz más sencilla. Prepara el grafo, busca ciclos y devuelve un iterador:
from graphlib import TopologicalSorter
pasos = {
"enviar_correo": {"generar_informe"},
"generar_informe": {"consultar_base", "cargar_config"},
"consultar_base": {"cargar_config"},
"cargar_config": set(),
}
for paso in TopologicalSorter(pasos).static_order():
print(f"Ejecutando: {paso}")Esta opción es apropiada para scripts pequeños, validadores, planes de migración y procesos que deben ejecutarse deliberadamente de uno en uno.
Agregar nodos de forma incremental
También puedes crear un objeto vacío y registrar dependencias con add(). El primer argumento es el nodo y los demás son sus predecesores:
from graphlib import TopologicalSorter
ordenador = TopologicalSorter()
ordenador.add("build", "lint", "test")
ordenador.add("deploy", "build")
ordenador.add("lint")
ordenador.add("test", "install")
ordenador.add("install")
print(list(ordenador.static_order()))Las llamadas repetidas para el mismo nodo acumulan requisitos. Esto resulta útil cuando diferentes módulos registran partes de una canalización. Después de preparar el grafo, ya no debe modificarse.
Procesamiento incremental y paralelo
La principal ventaja de TopologicalSorter aparece cuando varias tareas independientes pueden ejecutarse al mismo tiempo. Los métodos importantes son prepare(), get_ready(), done() e is_active().
from graphlib import TopologicalSorter
pasos = {
"deploy": {"tests", "frontend_build"},
"tests": {"install"},
"frontend_build": {"install"},
"install": set(),
}
ordenador = TopologicalSorter(pasos)
ordenador.prepare()
while ordenador.is_active():
listas = ordenador.get_ready()
for tarea in listas:
print("Ejecutando", tarea)
# ejecutar y esperar el resultado
ordenador.done(tarea)Cuando install termina, tests y frontend_build quedan disponibles simultáneamente. Pueden enviarse a hilos, procesos, contenedores o trabajadores remotos. done() debe llamarse únicamente después de confirmar la finalización correcta.
Ejemplo con ThreadPoolExecutor
Para operaciones dominadas por entrada y salida, un grupo de hilos puede aprovechar el paralelismo:
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
from graphlib import TopologicalSorter
import time
pasos = {
"publicar": {"probar", "documentar"},
"probar": {"compilar"},
"documentar": {"compilar"},
"compilar": {"descargar"},
"descargar": set(),
}
def ejecutar(nombre):
print("inicio", nombre)
time.sleep(0.2)
print("fin", nombre)
return nombre
ordenador = TopologicalSorter(pasos)
ordenador.prepare()
futuros = {}
with ThreadPoolExecutor(max_workers=3) as executor:
while ordenador.is_active():
for tarea in ordenador.get_ready():
futuro = executor.submit(ejecutar, tarea)
futuros[futuro] = tarea
terminados, _ = wait(futuros, return_when=FIRST_COMPLETED)
for futuro in terminados:
tarea = futuros.pop(futuro)
futuro.result()
ordenador.done(tarea)Un sistema real debe definir qué hacer cuando una tarea falla. No llames a done() si el requisito terminó con error, salvo que el flujo permita continuar. Puedes detener el grafo, reintentar, omitir dependientes o guardar un resultado parcial.
Detectar dependencias circulares
Tanto prepare() como static_order() pueden lanzar CycleError. Los argumentos de la excepción incluyen información sobre un ciclo detectado:
from graphlib import CycleError, TopologicalSorter
grafo = {
"a": {"c"},
"b": {"a"},
"c": {"b"},
}
try:
print(list(TopologicalSorter(grafo).static_order()))
except CycleError as error:
print("Dependencia circular:", error.args)El objeto puede liberar nodos que están fuera del ciclo antes de quedar bloqueado. Una herramienta de diagnóstico puede mostrar la parte válida y señalar después el componente problemático.
Nodos predecesores implícitos
Si un predecesor aparece en un valor pero no existe como clave, graphlib lo agrega automáticamente como nodo sin requisitos. Es cómodo, pero puede ocultar errores de escritura. configuracion y configuracon se convierten en dos nodos distintos.
Valida los identificadores contra un catálogo autorizado:
catalogo = {"config", "extraer", "transformar", "cargar"}
dependencias = {
"transformar": {"extraer"},
"cargar": {"transformar", "config"},
}
referenciados = set(dependencias)
for predecesores in dependencias.values():
referenciados.update(predecesores)
desconocidos = referenciados - catalogo
if desconocidos:
raise ValueError(f"Tareas desconocidas: {desconocidos}")Usar objetos como nodos
Los nodos deben ser hashable, pero no tienen que ser cadenas. Las enumeraciones, tuplas y clases de datos inmutables permiten transportar metadatos:
from dataclasses import dataclass
from graphlib import TopologicalSorter
@dataclass(frozen=True)
class Tarea:
nombre: str
equipo: str
extraer = Tarea("extraer", "datos")
validar = Tarea("validar", "calidad")
publicar = Tarea("publicar", "plataforma")
grafo = {
validar: {extraer},
publicar: {validar},
}
for tarea in TopologicalSorter(grafo).static_order():
print(tarea.nombre, tarea.equipo)Evita objetos mutables cuyo hash o igualdad pueda cambiar después de insertarlos, porque los diccionarios y conjuntos dejarían de comportarse de forma predecible.
Aplicaciones prácticas
La ordenación topológica aparece en:
- resolución de dependencias de paquetes y plugins;
- canalizaciones de datos y aprendizaje automático;
- migraciones de bases de datos;
- compilación de módulos;
- planificación de cursos con prerrequisitos;
- generación de informes y hojas dependientes;
- secuencia de arranque de servicios;
- sistemas de build y despliegue.
El módulo determina el orden y la disponibilidad. No incluye persistencia, colas distribuidas, límites de recursos, reintentos automáticos, horarios ni monitorización. Puede actuar como núcleo de planificación dentro de un sistema más completo.
Cómo probar el grafo
Cuando existen varias órdenes válidas, comprobar una lista exacta crea una prueba frágil. Es mejor verificar la propiedad esencial: cada predecesor aparece antes que su dependiente.
from graphlib import TopologicalSorter
def validar_orden(grafo):
orden = list(TopologicalSorter(grafo).static_order())
posicion = {nodo: indice for indice, nodo in enumerate(orden)}
for nodo, predecesores in grafo.items():
for anterior in predecesores:
assert posicion[anterior] < posicion[nodo]
validar_orden({
"deploy": {"test", "build"},
"test": {"install"},
"build": {"install"},
"install": set(),
})Añade pruebas para un grafo vacío, un nodo aislado, varias raíces, ciclos directos e indirectos, registros repetidos, identificadores desconocidos y fallos de trabajadores.
Errores frecuentes
- Proporcionar sucesores en lugar de predecesores.
- Suponer que las tareas independientes siempre mantienen el mismo orden.
- Olvidar
done()en el procesamiento incremental. - Marcar una tarea como terminada antes de que realmente tenga éxito.
- Modificar el grafo después de prepararlo.
- Ignorar excepciones y liberar dependientes incorrectamente.
- Usar nombres inconsistentes que crean nodos inesperados.
- Esperar que graphlib ejecute las tareas por sí mismo.
Buenas prácticas
- Usa identificadores estables e inmutables.
- Valida nodos desconocidos antes de preparar.
- Documenta que el mapa va de nodos a predecesores.
- Separa planificación, ejecución y persistencia.
- Define políticas de error y reintento antes de paralelizar.
- Limita la concurrencia según CPU, memoria y servicios externos.
- Registra inicio, fin, duración y resultado.
- Prueba propiedades del grafo, no desempates arbitrarios.
Conclusión
graphlib en Python convierte relaciones de dependencia en un plan ejecutable sin instalar paquetes externos. static_order() cubre los casos secuenciales, mientras que prepare(), get_ready(), done() e is_active() permiten coordinar ejecución concurrente.
La parte fundamental es modelar correctamente los predecesores y tratar los ciclos como errores de configuración. TopologicalSorter no sustituye a un orquestador completo, pero proporciona una base fiable para instaladores, pipelines, migraciones, compiladores y cualquier proceso gobernado por prerrequisitos.
Para consultar detalles de cada versión, revisa la documentación oficial de graphlib y la documentación de concurrent.futures.
