Function build_dag

src/zenml/utils/run_utils.py:170–185

Build DAG with downstream steps from a list of steps. Args: steps: The list of steps. Returns: The DAG with downstream steps.

(steps: Dict[str, List[str]])

def build_dag(steps: Dict[str, List[str]]) -> Dict[str, Set[str]]:
    """Build DAG with downstream steps from a list of steps.

    Args:
        steps: The list of steps.

    Returns:
        The DAG with downstream steps.
    """
    dag: Dict[str, Set[str]] = {step: set() for step in steps}

    for step_name, upstream_steps in steps.items():
        for upstream_step in upstream_steps:
            dag[upstream_step].add(step_name)

    return dag


def find_all_downstream_steps(