Pythonで依存関係に基づくタスクスケジューラを実装する

依存関係ベースのタスクスケジューラ。

実行プランをバッチ（タスクIDのリスト）のリストとして生成します。
各バッチ内のすべてのタスクは並列実行可能であり、バッチは順次実行する必要があります。
Kahn のアルゴリズム（レベルごとのトポロジカルソート）を使用します。

以下の場合に ValueError を発生させます。

• タスクIDの重複
• 未知のタスクIDを参照する依存関係
• 循環依存（問題のあるタスクをリスト化）

from collections import deque
from typing import Dict, List


class TaskScheduler:
    """依存関係に基づいてタスクを並列化可能なバッチにスケジュールします。"""

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

    def _validate_and_build_graph(self) -> None:
        # すべてのタスクIDを収集し、重複をチェックします。
        self.task_ids = set()
        for task in self.tasks:
            if 'id' not in task:
                raise ValueError(f"タスクに必要な 'id' キーがありません: {task!r}")
            task_id = task['id']
            if task_id in self.task_ids:
                raise ValueError(f"重複するタスクIDが見つかりました: '{task_id}'")
            self.task_ids.add(task_id)

        # 隣接リスト（依存関係 -> 依存されるもの）と入次数マップを構築します。
        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"タスク '{task_id}' は未知のタスクID '{dep}' に依存しています"
                    )
                if dep == task_id:
                    raise ValueError(
                        f"タスク '{task_id}' はそれ自体に依存しています（循環依存）"
                    )
                self.dependents[dep].append(task_id)
                self.in_degree[task_id] += 1

    def schedule(self) -> List[List[str]]:
        """
        実行プランをバッチのリストとして返します。

        各バッチは並列実行可能なタスクIDのリストです。
        バッチは返された順序で実行する必要があります。
        """
        in_degree = dict(self.in_degree)  # コピーで作業します
        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)  # 決定論的な出力（順序は任意）
            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(
                "循環依存が検出されました。関連タスク: "
                + ", ".join(f"'{t}'" for t in unresolved)
            )

        return plan


def schedule_tasks(tasks: List[dict]) -> List[List[str]]:
    """
    便利な関数：タスク辞書のリストをスケジュールします。

    Args:
        tasks: タスクの辞書のリスト。各辞書にはキー 'id' (str)、オプションで 'name'
               (str)、および 'dependencies' (タスクID文字列のリスト) が含まれます。

    Returns:
        タスクIDのリストのリスト。各内部リストは並列実行可能なタスクのバッチです。
        バッチは指定された順序で順次実行されます。

    Raises:
        ValueError: 依存関係が存在しないタスクIDを参照している場合、
                    循環依存が存在する場合、またはタスクIDが重複している場合。
    """
    return TaskScheduler(tasks).schedule()


if __name__ == '__main__':
    # 例 1
    tasks = [
        {'id': 'A', 'name': 'データ取り込み', 'dependencies': []},
        {'id': 'B', 'name': 'データクリーニング', 'dependencies': ['A']},
        {'id': 'C', 'name': '特徴量エンジニアリング', 'dependencies': ['A']},
        {'id': 'D', 'name': 'モデルトレーニング', 'dependencies': ['B', 'C']},
        {'id': 'E', 'name': 'ログ設定', 'dependencies': []},
        {'id': 'F', 'name': 'モデルデプロイ', 'dependencies': ['D']},
    ]
    print(schedule_tasks(tasks))
    # -> [['A', 'E'], ['B', 'C'], ['D'], ['F']]

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

    # 例 3: 循環依存は ValueError を発生させます
    cyclic = [
        {'id': 'X', 'dependencies': ['Y']},
        {'id': 'Y', 'dependencies': ['X']},
    ]
    try:
        schedule_tasks(cyclic)
    except ValueError as e:
        print(f"ValueError: {e}")

    # 例 4: 未知の依存関係は 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]]:
"""
タスクのリストに基づいて依存関係のある実行計画を構築します。

返される各内部リストは、並行して実行できるタスクIDのバッチです。
バッチは返された順序で実行する必要があります。

Args:
    tasks: 辞書のリスト。各タスクには以下が含まれている必要があります。
        - 'id': 一意の文字列タスク識別子
        - 'dependencies': 先に完了する必要があるタスクIDのリスト

Returns:
    バッチのリスト。各バッチはタスクIDのリストです。

Raises:
    ValueError: タスクIDが重複している場合、依存関係が不正な形式の場合、
                依存関係が不明なタスクを参照している場合、またはサイクルが存在する場合。
"""
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 は、各タスクが完了した後に実行可能に近づくタスクをマッピングします。
adjacency: Dict[str, List[str]] = {task_id: [] for task_id in task_ids}

# in_degree は、各タスクが持つ未完了の依存関係の数をカウントします。
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", [])

    # 単一タスクでの重複依存関係IDは、あいまいなブックキーピングを作成し、通常はデータエラーを示すため、無効な入力として扱います。
    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:
"""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))
# 例: [['A', 'E'], ['B', 'C'], ['D'], ['F']]
# 各バッチ内の順序は異なる場合があります。バッチ内のタスクは並行して実行されるためです。