Implemente um Agendador de Tarefas Baseado em Dependências em Python

Agendador de tarefas baseado em dependências.

Produz um plano de execução como uma lista de lotes (listas de IDs de tarefas),
onde cada tarefa em um lote pode ser executada simultaneamente, e os lotes devem ser executados sequencialmente.
Usa o algoritmo de Kahn (ordenação topológica nível a nível).

Levanta ValueError em caso de:

• IDs de tarefas duplicados,
• dependências que referenciam IDs de tarefas desconhecidos,
• dependências circulares (com as tarefas problemáticas listadas).

from collections import deque
from typing import Dict, List


class TaskScheduler:
    """Agenda tarefas em lotes paralelizados com base em dependências."""

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

    def _validate_and_build_graph(self) -> None:
        # Coleta todos os IDs de tarefas e verifica duplicatas.
        self.task_ids = set()
        for task in self.tasks:
            if 'id' not in task:
                raise ValueError(f"Tarefa está sem a chave 'id' obrigatória: {task!r}")
            task_id = task['id']
            if task_id in self.task_ids:
                raise ValueError(f"ID de tarefa duplicado encontrado: '{task_id}'")
            self.task_ids.add(task_id)

        # Constrói a lista de adjacências (dependência -> dependentes) e o mapa de grau 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"Tarefa '{task_id}' depende do ID de tarefa desconhecido '{dep}'"
                    )
                if dep == task_id:
                    raise ValueError(
                        f"Tarefa '{task_id}' depende de si mesma (dependência circular)"
                    )
                self.dependents[dep].append(task_id)
                self.in_degree[task_id] += 1

    def schedule(self) -> List[List[str]]:
        """
        Retorna o plano de execução como uma lista de lotes.

        Cada lote é uma lista de IDs de tarefas que podem ser executados simultaneamente.
        Os lotes devem ser executados na ordem retornada.
        """
        in_degree = dict(self.in_degree)  # trabalha em uma cópia
        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)  # saída determinística (ordem é arbitrária)
            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(
                "Dependência circular detectada envolvendo as tarefas: "
                + ", ".join(f"'{t}'" for t in unresolved)
            )

        return plan


def schedule_tasks(tasks: List[dict]) -> List[List[str]]:
    """
    Função de conveniência: agenda uma lista de dicionários de tarefas.

    Args:
        tasks: Lista de dicionários, cada um com as chaves 'id' (str), opcionalmente 'name'
               (str), e 'dependencies' (lista de strings de ID de tarefa).

    Returns:
        Uma lista de listas de IDs de tarefas. Cada lista interna é um lote de tarefas
        que podem ser executadas em paralelo; os lotes são executados sequencialmente em ordem.

    Raises:
        ValueError: Se uma dependência se refere a um ID de tarefa inexistente, se uma
                    dependência circular existir, ou se os IDs de tarefa forem duplicados.
    """
    return TaskScheduler(tasks).schedule()


if __name__ == '__main__':
    # Exemplo 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']]

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

    # Exemplo 3: dependência circular levanta ValueError
    cyclic = [
        {'id': 'X', 'dependencies': ['Y']},
        {'id': 'Y', 'dependencies': ['X']},
    ]
    try:
        schedule_tasks(cyclic)
    except ValueError as e:
        print(f"ValueError: {e}")

    # Exemplo 4: dependência desconhecida levanta 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]]:
"""
Constrói um plano de execução baseado em dependências para uma lista de tarefas.

Cada lista interna retornada é um lote de IDs de tarefas que podem ser executadas simultaneamente.
Os lotes devem ser executados na ordem retornada.

Args:
    tasks: Uma lista de dicionários. Cada tarefa deve conter:
        - 'id': identificador único da tarefa em formato de string
        - 'dependencies': lista de IDs de tarefas que devem ser concluídas primeiro

Returns:
    Uma lista de lotes, em que cada lote é uma lista de IDs de tarefas.

Raises:
    ValueError: Se os IDs das tarefas estiverem duplicados, as dependências estiverem malformadas,
                uma dependência referenciar uma tarefa desconhecida, ou existir um ciclo.
"""
if not isinstance(tasks, list):
    raise ValueError("tasks deve ser uma lista de dicionários de tarefas")

task_ids: Set[str] = set()

for task in tasks:
    if not isinstance(task, dict):
        raise ValueError("cada tarefa deve ser um dicionário")

    task_id = task.get("id")
    if not isinstance(task_id, str) or not task_id:
        raise ValueError("cada tarefa deve ter um 'id' de string não vazia")

    if task_id in task_ids:
        raise ValueError(f"id de tarefa duplicado encontrado: {task_id!r}")

    dependencies = task.get("dependencies", [])
    if not isinstance(dependencies, list):
        raise ValueError(f"as dependências da tarefa {task_id!r} devem ser uma lista")

    for dependency_id in dependencies:
        if not isinstance(dependency_id, str) or not dependency_id:
            raise ValueError(
                f"as dependências da tarefa {task_id!r} devem conter apenas strings não vazias"
            )

    task_ids.add(task_id)

# adjacency mapeia cada tarefa para as tarefas que ficam mais próximas de poder ser executadas após sua conclusão.
adjacency: Dict[str, List[str]] = {task_id: [] for task_id in task_ids}

# in_degree conta quantas dependências não concluídas cada tarefa tem.
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", [])

    # Trate IDs de dependência duplicados em uma única tarefa como uma entrada inválida porque,
    # caso contrário, eles criam um controle ambíguo e normalmente indicam um erro nos dados.
    seen_dependencies: Set[str] = set()

    for dependency_id in dependencies:
        if dependency_id not in task_ids:
            raise ValueError(
                f"a tarefa {task_id!r} depende de um id de tarefa desconhecido {dependency_id!r}"
            )

        if dependency_id in seen_dependencies:
            raise ValueError(
                f"a tarefa {task_id!r} lista a dependência {dependency_id!r} mais de uma 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(
        "dependência circular detectada envolvendo a(s) tarefa(s): "
        + ", ".join(cyclic_task_ids)
    )

return execution_plan

class TaskScheduler:
"""Pequeno wrapper de conveniência em torno 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))
# Saída de exemplo: [['A', 'E'], ['B', 'C'], ['D'], ['F']]
# A ordem dentro de cada lote pode diferir porque as tarefas em um lote são simultâneas.