Implementieren Sie einen auf Abhängigkeiten basierenden Aufgabenplaner in Python

Abhängigkeitsbasierter Aufgabendisponent.

Erzeugt einen Ausführungsplan als Liste von Batches (Listen von Aufgaben-IDs),
bei denen jede Aufgabe in einem Batch parallel ausgeführt werden kann und Batches
sequenziell ausgeführt werden müssen. Verwendet Kahns Algorithmus (topologische Sortierung auf Ebene für Ebene).

Löst ValueError aus bei:

• doppelten Aufgaben-IDs,
• Abhängigkeiten, die auf unbekannte Aufgaben-IDs verweisen,
• zirkulären Abhängigkeiten (mit den betroffenen Aufgaben aufgeführt).

from collections import deque
from typing import Dict, List

class TaskScheduler:
"""Plant Aufgaben basierend auf Abhängigkeiten in parallelisierbare Batches."""

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

def _validate_and_build_graph(self) -> None:
    # Sammelt alle Aufgaben-IDs und prüft auf Duplikate.
    self.task_ids = set()
    for task in self.tasks:
        if 'id' not in task:
            raise ValueError(f"Aufgabe hat den erforderlichen Schlüssel 'id' nicht: {task!r}")
        task_id = task['id']
        if task_id in self.task_ids:
            raise ValueError(f"Doppelte Aufgaben-ID gefunden: '{task_id}'")
        self.task_ids.add(task_id)

    # Erstellt Adjazenzliste (Abhängigkeit -> Abhängige) und In-Degree-Map.
    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"Aufgabe '{task_id}' hängt von unbekannter Aufgaben-ID '{dep}' ab"
                )
            if dep == task_id:
                raise ValueError(
                    f"Aufgabe '{task_id}' hängt von sich selbst ab (zirkuläre Abhängigkeit)"
                )
            self.dependents[dep].append(task_id)
            self.in_degree[task_id] += 1

def schedule(self) -> List[List[str]]:
    """
    Gibt den Ausführungsplan als Liste von Batches zurück.

    Jeder Batch ist eine Liste von Aufgaben-IDs, die parallel ausgeführt werden können.
    Batches müssen in der zurückgegebenen Reihenfolge ausgeführt werden.
    """
    in_degree = dict(self.in_degree)  # Arbeitet mit einer Kopie
    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)  # deterministische Ausgabe (Reihenfolge ist beliebig)
        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(
            "Zirkuläre Abhängigkeit erkannt, die Aufgaben betrifft: "
            + ", ".join(f"'{t}'" for t in unresolved)
        )

    return plan

def schedule_tasks(tasks: List[dict]) -> List[List[str]]:
"""
Hilfsfunktion: Plant eine Liste von Aufgaben-Dictionaries.

Args:
    tasks: Liste von Dictionaries, jedes mit den Schlüsseln 'id' (str),
           optional 'name' (str) und 'dependencies' (Liste von Aufgaben-ID-Strings).

Returns:
    Eine Liste von Listen von Aufgaben-IDs. Jede innere Liste ist ein Batch von Aufgaben,
    die parallel ausgeführt werden können; Batches werden sequenziell in der Reihenfolge ausgeführt.

Raises:
    ValueError: Wenn eine Abhängigkeit auf eine nicht existierende Aufgaben-ID verweist, wenn
                eine zirkuläre Abhängigkeit besteht oder wenn Aufgaben-IDs dupliziert werden.
"""
return TaskScheduler(tasks).schedule()

if name == 'main':
# Beispiel 1
tasks = [
{'id': 'A', 'name': 'Datenerfassung', 'dependencies': []},
{'id': 'B', 'name': 'Datenbereinigung', 'dependencies': ['A']},
{'id': 'C', 'name': 'Feature Engineering', 'dependencies': ['A']},
{'id': 'D', 'name': 'Modelltraining', 'dependencies': ['B', 'C']},
{'id': 'E', 'name': 'Logging einrichten', 'dependencies': []},
{'id': 'F', 'name': 'Modellbereitstellung', 'dependencies': ['D']},
]
print(schedule_tasks(tasks))
# -> [['A', 'E'], ['B', 'C'], ['D'], ['F']]

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

# Beispiel 3: zirkuläre Abhängigkeit löst ValueError aus
cyclic = [
    {'id': 'X', 'dependencies': ['Y']},
    {'id': 'Y', 'dependencies': ['X']},
]
try:
    schedule_tasks(cyclic)
except ValueError as e:
    print(f"ValueError: {e}")

# Beispiel 4: unbekannte Abhängigkeit löst ValueError aus
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]]:
"""
Erstelle einen abhängigkeitbasierten Ausführungsplan für eine Liste von Aufgaben.

Jede zurückgegebene innere Liste ist ein Stapel von Aufgaben-IDs, die gleichzeitig ausgeführt werden können.
Stapel müssen in der zurückgegebenen Reihenfolge ausgeführt werden.

Argumente:
    tasks: Eine Liste von Dictionaries. Jede Aufgabe muss enthalten:
        - 'id': eindeutiger String-Aufgabenbezeichner
        - 'dependencies': Liste von Aufgaben-IDs, die zuerst abgeschlossen werden müssen

Rückgabewert:
    Eine Liste von Stapeln, wobei jeder Stapel eine Liste von Aufgaben-IDs ist.

Löst aus:
    ValueError: Wenn Aufgaben-IDs doppelt vorkommen, Abhängigkeiten fehlerhaft formatiert sind,
                eine Abhängigkeit auf eine unbekannte Aufgabe verweist oder ein Zyklus existiert.
"""
if not isinstance(tasks, list):
    raise ValueError("tasks muss eine Liste von Aufgaben-Dictionaries sein")

task_ids: Set[str] = set()

for task in tasks:
    if not isinstance(task, dict):
        raise ValueError("jede Aufgabe muss ein Dictionary sein")

    task_id = task.get("id")
    if not isinstance(task_id, str) or not task_id:
        raise ValueError("jede Aufgabe muss eine nicht-leere String-'id' haben")

    if task_id in task_ids:
        raise ValueError(f"doppelte Aufgaben-ID gefunden: {task_id!r}")

    dependencies = task.get("dependencies", [])
    if not isinstance(dependencies, list):
        raise ValueError(f"Abhängigkeiten für Aufgabe {task_id!r} müssen eine Liste sein")

    for dependency_id in dependencies:
        if not isinstance(dependency_id, str) or not dependency_id:
            raise ValueError(
                f"Abhängigkeiten für Aufgabe {task_id!r} dürfen nur nicht-leere Strings enthalten"
            )

    task_ids.add(task_id)

# adjacency ordnet jeder Aufgabe die Aufgaben zu, die nach ihrem Abschluss ausführbarer werden.
adjacency: Dict[str, List[str]] = {task_id: [] for task_id in task_ids}

# in_degree zählt, wie viele unerledigte Abhängigkeiten jede Aufgabe hat.
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", [])

    # Behandle doppelte Abhängigkeits-IDs innerhalb einer einzelnen Aufgabe als ungültige Eingabe, da sie
    # sonst zu uneindeutiger Buchführung führen und normalerweise auf einen Datenfehler hinweisen.
    seen_dependencies: Set[str] = set()

    for dependency_id in dependencies:
        if dependency_id not in task_ids:
            raise ValueError(
                f"Aufgabe {task_id!r} hängt von unbekannter Aufgaben-ID {dependency_id!r} ab"
            )

        if dependency_id in seen_dependencies:
            raise ValueError(
                f"Aufgabe {task_id!r} führt die Abhängigkeit {dependency_id!r} mehr als einmal auf"
            )
        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(
        "zirkuläre Abhängigkeit erkannt mit Aufgabe(n): "
        + ", ".join(cyclic_task_ids)
    )

return execution_plan

class TaskScheduler:
"""Kleine praktische Hülle um 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": "Datenaufnahme", "dependencies": []},
{"id": "B", "name": "Datenbereinigung", "dependencies": ["A"]},
{"id": "C", "name": "Feature-Engineering", "dependencies": ["A"]},
{"id": "D", "name": "Modelltraining", "dependencies": ["B", "C"]},
{"id": "E", "name": "Logging einrichten", "dependencies": []},
{"id": "F", "name": "Modellbereitstellung", "dependencies": ["D"]},
]

print(schedule_tasks(tasks))
# Beispielausgabe: [['A', 'E'], ['B', 'C'], ['D'], ['F']]
# Die Reihenfolge innerhalb jedes Stapels kann abweichen, da Aufgaben in einem Stapel gleichzeitig laufen können.