Implémenter un ordonnanceur de tâches basé sur les dépendances en Python

Planificateur de tâches basé sur les dépendances.

Produit un plan d'exécution sous forme de liste de lots (listes d'identifiants de tâches), où chaque tâche d'un lot peut s'exécuter simultanément, et les lots doivent s'exécuter séquentiellement. Utilise l'algorithme de Kahn (tri topologique niveau par niveau).

Lève une ValueError en cas de :

• identifiants de tâches dupliqués,
• dépendances faisant référence à des identifiants de tâches inconnus,
• dépendances circulaires (avec les tâches fautives listées).

from collections import deque
from typing import Dict, List

class TaskScheduler:
"""Planifie les tâches en lots parallélisables en fonction des dépendances."""

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

def _validate_and_build_graph(self) -> None:
    # Collecte tous les identifiants de tâches et vérifie les doublons.
    self.task_ids = set()
    for task in self.tasks:
        if 'id' not in task:
            raise ValueError(f"La tâche est manquante de la clé 'id' requise : {task!r}")
        task_id = task['id']
        if task_id in self.task_ids:
            raise ValueError(f"Identifiant de tâche dupliqué trouvé : '{task_id}'")
        self.task_ids.add(task_id)

    # Construit la liste d'adjacence (dépendance -> dépendants) et la carte des degrés entrants.
    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 tâche '{task_id}' dépend de l'identifiant de tâche inconnu '{dep}'"
                )
            if dep == task_id:
                raise ValueError(
                    f"La tâche '{task_id}' dépend d'elle-même (dépendance circulaire)"
                )
            self.dependents[dep].append(task_id)
            self.in_degree[task_id] += 1

def schedule(self) -> List[List[str]]:
    """
    Retourne le plan d'exécution sous forme de liste de lots.

    Chaque lot est une liste d'identifiants de tâches qui peuvent être exécutés simultanément.
    Les lots doivent être exécutés dans l'ordre retourné.
    """
    in_degree = dict(self.in_degree)  # travaille sur une copie
    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)  # sortie déterministe (l'ordre est arbitraire)
        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(
            "Dépendance circulaire détectée impliquant les tâches : "
            + ", ".join(f"'{t}'" for t in unresolved)
        )

    return plan

def schedule_tasks(tasks: List[dict]) -> List[List[str]]:
"""
Fonction de commodité : planifie une liste de dictionnaires de tâches.

Args:
    tasks: Liste de dictionnaires, chacun avec les clés 'id' (str), éventuellement 'name'
           (str), et 'dependencies' (liste de chaînes d'identifiants de tâches).

Returns:
    Une liste de listes d'identifiants de tâches. Chaque liste interne est un lot de tâches
    qui peuvent s'exécuter en parallèle ; les lots s'exécutent séquentiellement dans l'ordre.

Raises:
    ValueError: Si une dépendance fait référence à un identifiant de tâche inexistant, si une
                dépendance circulaire existe, ou si les identifiants de tâches sont dupliqués.
"""
return TaskScheduler(tasks).schedule()

if name == 'main':
# Exemple 1
tasks = [
{'id': 'A', 'name': 'Ingestion de données', 'dependencies': []},
{'id': 'B', 'name': 'Nettoyage de données', 'dependencies': ['A']},
{'id': 'C', 'name': 'Ingénierie des caractéristiques', 'dependencies': ['A']},
{'id': 'D', 'name': 'Entraînement du modèle', 'dependencies': ['B', 'C']},
{'id': 'E', 'name': 'Configuration de la journalisation', 'dependencies': []},
{'id': 'F', 'name': 'Déploiement du modèle', 'dependencies': ['D']},
]
print(schedule_tasks(tasks))
# -> [['A', 'E'], ['B', 'C'], ['D'], ['F']]

# Exemple 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']]

# Exemple 3 : la dépendance circulaire lève une ValueError
cyclic = [
    {'id': 'X', 'dependencies': ['Y']},
    {'id': 'Y', 'dependencies': ['X']},
]
try:
    schedule_tasks(cyclic)
except ValueError as e:
    print(f"ValueError: {e}")

# Exemple 4 : la dépendance inconnue lève une 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]]:
"""
Construit un plan d’exécution basé sur les dépendances pour une liste de tâches.

Chaque liste interne renvoyée est un lot d’identifiants de tâches pouvant être exécutées simultanément.
Les lots doivent être exécutés dans l’ordre renvoyé.

Args:
    tasks: Une liste de dictionnaires. Chaque tâche doit contenir :
        - 'id' : identifiant unique de tâche sous forme de chaîne
        - 'dependencies' : liste des identifiants de tâches qui doivent d’abord être terminées

Returns:
    Une liste de lots, où chaque lot est une liste d’identifiants de tâches.

Raises:
    ValueError: Si des identifiants de tâches sont dupliqués, si les dépendances sont mal formées,
                si une dépendance fait référence à une tâche inconnue, ou s’il existe un cycle.
"""
if not isinstance(tasks, list):
    raise ValueError("tasks must be a list of task dictionaries")

task_ids: Set[str] = set()

for task in tasks:
    if not isinstance(task, dict):
        raise ValueError("each task must be a dictionary")

    task_id = task.get("id")
    if not isinstance(task_id, str) or not task_id:
        raise ValueError("each task must have a non-empty string 'id'")

    if task_id in task_ids:
        raise ValueError(f"duplicate task id found: {task_id!r}")

    dependencies = task.get("dependencies", [])
    if not isinstance(dependencies, list):
        raise ValueError(f"dependencies for task {task_id!r} must be a list")

    for dependency_id in dependencies:
        if not isinstance(dependency_id, str) or not dependency_id:
            raise ValueError(
                f"dependencies for task {task_id!r} must contain only non-empty strings"
            )

    task_ids.add(task_id)

# adjacency associe chaque tâche aux tâches qui deviennent plus proches de pouvoir être exécutées après son achèvement.
adjacency: Dict[str, List[str]] = {task_id: [] for task_id in task_ids}

# in_degree compte combien de dépendances non terminées chaque tâche possède.
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", [])

    # Considérer des identifiants de dépendance dupliqués sur une même tâche comme une entrée invalide, car ils
    # créent sinon un suivi ambigu et indiquent généralement une erreur de données.
    seen_dependencies: Set[str] = set()

    for dependency_id in dependencies:
        if dependency_id not in task_ids:
            raise ValueError(
                f"task {task_id!r} depends on unknown task id {dependency_id!r}"
            )

        if dependency_id in seen_dependencies:
            raise ValueError(
                f"task {task_id!r} lists dependency {dependency_id!r} more than once"
            )
        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(
        "circular dependency detected involving task(s): "
        + ", ".join(cyclic_task_ids)
    )

return execution_plan

class TaskScheduler:
"""Petit wrapper utilitaire autour 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": "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))
# Exemple de sortie : [['A', 'E'], ['B', 'C'], ['D'], ['F']]
# L’ordre au sein de chaque lot peut différer, car les tâches d’un lot sont concurrentes.