Implement a Dependency-Based Task Scheduler in Python

"""
Dependency-based task scheduler.

Produces an execution plan as a list of batches (lists of task IDs),
where every task in a batch can run concurrently, and batches must run
sequentially. Uses Kahn's algorithm (level-by-level topological sort).

Raises ValueError on:

• duplicate task IDs,
• dependencies that reference unknown task IDs,
• circular dependencies (with the offending tasks listed).
  """

from collections import deque
from typing import Dict, List

class TaskScheduler:
"""Schedules tasks into parallelizable batches based on dependencies."""

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

def _validate_and_build_graph(self) -> None:
    # Collect all task IDs and check for duplicates.
    self.task_ids = set()
    for task in self.tasks:
        if 'id' not in task:
            raise ValueError(f"Task is missing required 'id' key: {task!r}")
        task_id = task['id']
        if task_id in self.task_ids:
            raise ValueError(f"Duplicate task ID found: '{task_id}'")
        self.task_ids.add(task_id)

    # Build adjacency list (dependency -> dependents) and 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"Task '{task_id}' depends on unknown task ID '{dep}'"
                )
            if dep == task_id:
                raise ValueError(
                    f"Task '{task_id}' depends on itself (circular dependency)"
                )
            self.dependents[dep].append(task_id)
            self.in_degree[task_id] += 1

def schedule(self) -> List[List[str]]:
    """
    Return the execution plan as a list of batches.

    Each batch is a list of task IDs that can be executed concurrently.
    Batches must be executed in the returned order.
    """
    in_degree = dict(self.in_degree)  # work on a copy
    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)  # deterministic output (order is arbitrary)
        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(
            "Circular dependency detected involving tasks: "
            + ", ".join(f"'{t}'" for t in unresolved)
        )

    return plan

def schedule_tasks(tasks: List[dict]) -> List[List[str]]:
"""
Convenience function: schedule a list of task dictionaries.

Args:
    tasks: List of dicts, each with keys 'id' (str), optionally 'name'
           (str), and 'dependencies' (list of task ID strings).

Returns:
    A list of lists of task IDs. Each inner list is a batch of tasks
    that can run in parallel; batches run sequentially in order.

Raises:
    ValueError: If a dependency refers to a non-existent task ID, if a
                circular dependency exists, or if task IDs are duplicated.
"""
return TaskScheduler(tasks).schedule()

if name == 'main':
# Example 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']]

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

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

# Example 4: unknown dependency raises 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]]:
"""
Build a dependency-based execution plan for a list of tasks.

Each returned inner list is a batch of task IDs that can be run concurrently.
Batches must be executed in the order returned.

Args:
    tasks: A list of dictionaries. Each task must contain:
        - 'id': unique string task identifier
        - 'dependencies': list of task IDs that must complete first

Returns:
    A list of batches, where each batch is a list of task IDs.

Raises:
    ValueError: If task IDs are duplicated, dependencies are malformed,
                a dependency references an unknown task, or a cycle exists.
"""
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 maps each task to the tasks that become closer to runnable after it completes.
adjacency: Dict[str, List[str]] = {task_id: [] for task_id in task_ids}

# in_degree counts how many unfinished dependencies each task has.
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", [])

    # Treat duplicate dependency IDs on a single task as an invalid input because they
    # otherwise create ambiguous bookkeeping and usually indicate a data error.
    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:
"""Small convenience wrapper around 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))
# Example output: [['A', 'E'], ['B', 'C'], ['D'], ['F']]
# The order within each batch may differ because tasks in a batch are concurrent.