Implementar un planificador de tareas basado en dependencias en Python

Planificador de tareas basado en dependencias.

Produce un plan de ejecución como una lista de lotes (listas de IDs de tareas),
donde cada tarea en un lote puede ejecutarse de forma concurrente, y los lotes deben ejecutarse secuencialmente.
Utiliza el algoritmo de Kahn (ordenación topológica nivel por nivel).

Genera ValueError en caso de:

• IDs de tareas duplicados,
• dependencias que hagan referencia a IDs de tareas desconocidos,
• dependencias circulares (con las tareas infractoras listadas).

from collections import deque
from typing import Dict, List

class TaskScheduler:
"""Planifica tareas en lotes paralelizados basándose en dependencias."""

def __init__(self, tasks: List[dict]):
    self.tasks = tasks
    self._validate_and_build_graph()

def _validate_and_build_graph(self) -> None:
    # Recopila todos los IDs de tareas y comprueba duplicados.
    self.task_ids = set()
    for task in self.tasks:
        if 'id' not in task:
            raise ValueError(f"La tarea carece de la clave 'id' requerida: {task!r}")
        task_id = task['id']
        if task_id in self.task_ids:
            raise ValueError(f"Se encontró un ID de tarea duplicado: '{task_id}'")
        self.task_ids.add(task_id)

    # Construye la lista de adyacencia (dependencia -> dependientes) y el mapa de grado de entrada.
    self.dependents: Dict[str, List[str]] = {tid: [] for tid in self.task_ids}
    self.in_degree: Dict[str, int] = {tid: 0 for tid in self.task_ids}

    for task in self.tasks:
        task_id = task['id']
        for dep in task.get('dependencies') or []:
            if dep not in self.task_ids:
                raise ValueError(
                    f"La tarea '{task_id}' depende de un ID de tarea desconocido '{dep}'"
                )
            if dep == task_id:
                raise ValueError(
                    f"La tarea '{task_id}' depende de sí misma (dependencia circular)"
                )
            self.dependents[dep].append(task_id)
            self.in_degree[task_id] += 1

def schedule(self) -> List[List[str]]:
    """
    Devuelve el plan de ejecución como una lista de lotes.

    Cada lote es una lista de IDs de tareas que se pueden ejecutar de forma concurrente.
    Los lotes deben ejecutarse en el orden devuelto.
    """
    in_degree = dict(self.in_degree)  # trabaja sobre una copia
    current_batch = deque(
        tid for tid in in_degree if in_degree[tid] == 0
    )

    plan: List[List[str]] = []
    scheduled_count = 0

    while current_batch:
        batch = sorted(current_batch)  # salida determinista (el orden es arbitrario)
        plan.append(batch)
        scheduled_count += len(batch)

        next_batch = deque()
        for task_id in batch:
            for dependent in self.dependents[task_id]:
                in_degree[dependent] -= 1
                if in_degree[dependent] == 0:
                    next_batch.append(dependent)
        current_batch = next_batch

    if scheduled_count != len(self.task_ids):
        unresolved = sorted(
            tid for tid in in_degree if in_degree[tid] > 0
        )
        raise ValueError(
            "Se detectó una dependencia circular que involucra a las tareas: "
            + ", ".join(f"'{t}'" for t in unresolved)
        )

    return plan

def schedule_tasks(tasks: List[dict]) -> List[List[str]]:
"""
Función de conveniencia: planifica una lista de diccionarios de tareas.

Args:
    tasks: Lista de diccionarios, cada uno con las claves 'id' (str),
           opcionalmente 'name' (str), y 'dependencies' (lista de strings de IDs de tareas).

Returns:
    Una lista de listas de IDs de tareas. Cada lista interna es un lote de tareas
    que se pueden ejecutar en paralelo; los lotes se ejecutan secuencialmente en orden.

Raises:
    ValueError: Si una dependencia hace referencia a un ID de tarea inexistente, si
                existe una dependencia circular, o si los IDs de las tareas están duplicados.
"""
return TaskScheduler(tasks).schedule()

if name == 'main':
# Ejemplo 1
tasks = [
{'id': 'A', 'name': 'Data Ingestion', 'dependencies': []},
{'id': 'B', 'name': 'Data Cleaning', 'dependencies': ['A']},
{'id': 'C', 'name': 'Feature Engineering', 'dependencies': ['A']},
{'id': 'D', 'name': 'Model Training', 'dependencies': ['B', 'C']},
{'id': 'E', 'name': 'Setup Logging', 'dependencies': []},
{'id': 'F', 'name': 'Model Deployment', 'dependencies': ['D']},
]
print(schedule_tasks(tasks))
# -> [['A', 'E'], ['B', 'C'], ['D'], ['F']]

# Ejemplo 2
pipeline = [
    {'id': 'checkout', 'dependencies': []},
    {'id': 'build', 'dependencies': ['checkout']},
    {'id': 'unit_test', 'dependencies': ['build']},
    {'id': 'integration_test', 'dependencies': ['build']},
    {'id': 'deploy', 'dependencies': ['unit_test', 'integration_test']},
]
print(schedule_tasks(pipeline))
# -> [['checkout'], ['build'], ['integration_test', 'unit_test'], ['deploy']]

# Ejemplo 3: dependencia circular genera ValueError
cyclic = [
    {'id': 'X', 'dependencies': ['Y']},
    {'id': 'Y', 'dependencies': ['X']},
]
try:
    schedule_tasks(cyclic)
except ValueError as e:
    print(f"ValueError: {e}")

# Ejemplo 4: dependencia desconocida genera ValueError
broken = [
    {'id': 'A', 'dependencies': ['missing_task']},
]
try:
    schedule_tasks(broken)
except ValueError as e:
    print(f"ValueError: {e}")

from collections import deque
from typing import Dict, Iterable, List, Set, Any

def schedule_tasks(tasks: List[Dict[str, Any]]) -> List[List[str]]:
"""
Construye un plan de ejecución basado en dependencias para una lista de tareas.

Cada lista interna devuelta es un lote de identificadores de tareas que pueden ejecutarse de forma concurrente.
Los lotes deben ejecutarse en el orden devuelto.

Argumentos:
    tasks: Una lista de diccionarios. Cada tarea debe contener:
        - 'id': identificador único de tarea en formato de cadena
        - 'dependencies': lista de identificadores de tareas que deben completarse primero

Devuelve:
    Una lista de lotes, donde cada lote es una lista de identificadores de tareas.

Lanza:
    ValueError: Si los identificadores de tarea están duplicados, las dependencias tienen un formato incorrecto,
                una dependencia hace referencia a una tarea desconocida, o existe un ciclo.
"""
if not isinstance(tasks, list):
    raise ValueError("tasks debe ser una lista de diccionarios de tareas")

task_ids: Set[str] = set()

for task in tasks:
    if not isinstance(task, dict):
        raise ValueError("cada tarea debe ser un diccionario")

    task_id = task.get("id")
    if not isinstance(task_id, str) or not task_id:
        raise ValueError("cada tarea debe tener un 'id' de cadena no vacía")

    if task_id in task_ids:
        raise ValueError(f"se encontró un id de tarea duplicado: {task_id!r}")

    dependencies = task.get("dependencies", [])
    if not isinstance(dependencies, list):
        raise ValueError(f"las dependencias de la tarea {task_id!r} deben ser una lista")

    for dependency_id in dependencies:
        if not isinstance(dependency_id, str) or not dependency_id:
            raise ValueError(
                f"las dependencias de la tarea {task_id!r} deben contener solo cadenas no vacías"
            )

    task_ids.add(task_id)

# adjacency asigna cada tarea a las tareas que pasan a estar más cerca de poder ejecutarse después de que se complete.
adjacency: Dict[str, List[str]] = {task_id: [] for task_id in task_ids}

# in_degree cuenta cuántas dependencias sin finalizar tiene cada tarea.
in_degree: Dict[str, int] = {task_id: 0 for task_id in task_ids}

for task in tasks:
    task_id = task["id"]
    dependencies = task.get("dependencies", [])

    # Tratar los identificadores de dependencia duplicados en una sola tarea como una entrada no válida porque,
    # de lo contrario, crean un seguimiento ambiguo y normalmente indican un error en los datos.
    seen_dependencies: Set[str] = set()

    for dependency_id in dependencies:
        if dependency_id not in task_ids:
            raise ValueError(
                f"la tarea {task_id!r} depende del id de tarea desconocido {dependency_id!r}"
            )

        if dependency_id in seen_dependencies:
            raise ValueError(
                f"la tarea {task_id!r} enumera la dependencia {dependency_id!r} más de una vez"
            )
        seen_dependencies.add(dependency_id)

        adjacency[dependency_id].append(task_id)
        in_degree[task_id] += 1

ready = deque(task_id for task_id in task_ids if in_degree[task_id] == 0)
execution_plan: List[List[str]] = []
scheduled_count = 0

while ready:
    batch = list(ready)
    ready.clear()
    execution_plan.append(batch)
    scheduled_count += len(batch)

    for completed_task_id in batch:
        for dependent_task_id in adjacency[completed_task_id]:
            in_degree[dependent_task_id] -= 1
            if in_degree[dependent_task_id] == 0:
                ready.append(dependent_task_id)

if scheduled_count != len(task_ids):
    cyclic_task_ids = sorted(
        task_id for task_id, remaining_dependencies in in_degree.items()
        if remaining_dependencies > 0
    )
    raise ValueError(
        "se detectó una dependencia circular que involucra la(s) tarea(s): "
        + ", ".join(cyclic_task_ids)
    )

return execution_plan

class TaskScheduler:
"""Pequeño contenedor de conveniencia alrededor de schedule_tasks()."""

def __init__(self, tasks: List[Dict[str, Any]]):
    self.tasks = tasks

def execution_plan(self) -> List[List[str]]:
    return schedule_tasks(self.tasks)

if name == "main":
tasks = [
{"id": "A", "name": "Ingesta de datos", "dependencies": []},
{"id": "B", "name": "Limpieza de datos", "dependencies": ["A"]},
{"id": "C", "name": "Ingeniería de características", "dependencies": ["A"]},
{"id": "D", "name": "Entrenamiento del modelo", "dependencies": ["B", "C"]},
{"id": "E", "name": "Configurar registro", "dependencies": []},
{"id": "F", "name": "Despliegue del modelo", "dependencies": ["D"]},
]

print(schedule_tasks(tasks))
# Salida de ejemplo: [['A', 'E'], ['B', 'C'], ['D'], ['F']]
# El orden dentro de cada lote puede variar porque las tareas de un lote son concurrentes.