Simplify and fix code in dag.py. (#418)

Author: tobiasraabe

docs/source/changes.md

@@ -35,6 +35,7 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and
 - {pull}`414` allow more ruff rules.
 - {pull}`416` removes `.from_annot` again.
 - {pull}`417` deprecates {func}`pytask.mark.task` in favor of {func}`pytask.task`.
+- {pull}`418` fixes and error and simplifies code in `dag.py`.
 
 ## 0.3.2 - 2023-06-07
 

pyproject.toml

@@ -64,6 +64,7 @@ extend-ignore = [
 "src/_pytask/capture.py" = ["PGH003"]
 "src/_pytask/hookspecs.py" = ["ARG001"]
 "src/_pytask/outcomes.py" = ["N818"]
+"src/_pytask/dag.py" = ["B023"]
 "tests/test_capture.py" = ["T201", "PT011"]
 "tests/*" = ["D", "ANN", "PLR2004", "S101"]
 "scripts/*" = ["D", "INP001"]

src/_pytask/dag.py

@@ -4,6 +4,7 @@
 import hashlib
 import itertools
 import sys
+from typing import Sequence
 from typing import TYPE_CHECKING
 
 import networkx as nx
@@ -38,6 +39,7 @@
 from rich.tree import Tree
 
 if TYPE_CHECKING:
+    from pathlib import Path
     from _pytask.session import Session
 
 
@@ -79,12 +81,10 @@ def pytask_dag_create_dag(tasks: list[PTask]) -> nx.DiGraph:
         dag.add_node(task.name, task=task)
 
         tree_map(lambda x: dag.add_node(x.name, node=x), task.depends_on)
-        tree_map(
-            lambda x: dag.add_edge(x.name, task.name), task.depends_on  # noqa: B023
-        )
+        tree_map(lambda x: dag.add_edge(x.name, task.name), task.depends_on)
 
         tree_map(lambda x: dag.add_node(x.name, node=x), task.produces)
-        tree_map(lambda x: dag.add_edge(task.name, x.name), task.produces)  # noqa: B023
+        tree_map(lambda x: dag.add_edge(task.name, x.name), task.produces)
 
     _check_if_dag_has_cycles(dag)
 
@@ -110,9 +110,9 @@ def pytask_dag_select_execution_dag(session: Session, dag: nx.DiGraph) -> None:
 
 
 @hookimpl
-def pytask_dag_validate_dag(dag: nx.DiGraph) -> None:
+def pytask_dag_validate_dag(session: Session, dag: nx.DiGraph) -> None:
     """Validate the DAG."""
-    _check_if_root_nodes_are_available(dag)
+    _check_if_root_nodes_are_available(dag, session.config["paths"])
     _check_if_tasks_have_the_same_products(dag)
 
 
@@ -200,7 +200,7 @@ def _format_cycles(cycles: list[tuple[str, ...]]) -> str:
     )
 
 
-def _check_if_root_nodes_are_available(dag: nx.DiGraph) -> None:
+def _check_if_root_nodes_are_available(dag: nx.DiGraph, paths: Sequence[Path]) -> None:
     missing_root_nodes = []
     is_task_skipped: dict[str, bool] = {}
 
@@ -217,23 +217,14 @@ def _check_if_root_nodes_are_available(dag: nx.DiGraph) -> None:
                     missing_root_nodes.append(node)
 
     if missing_root_nodes:
-        all_names = missing_root_nodes + [
-            successor
-            for node in missing_root_nodes
-            for successor in dag.successors(node)
-            if not is_task_skipped[successor]
-        ]
-        common_ancestor = find_common_ancestor_of_nodes(*all_names)
         dictionary = {}
         for node in missing_root_nodes:
-            short_node_name = reduce_node_name(
-                dag.nodes[node]["node"], [common_ancestor]
-            )
+            short_node_name = reduce_node_name(dag.nodes[node]["node"], paths)
             not_skipped_successors = [
                 task for task in dag.successors(node) if not is_task_skipped[task]
             ]
             short_successors = reduce_names_of_multiple_nodes(
-                not_skipped_successors, dag, [common_ancestor]
+                not_skipped_successors, dag, paths
             )
             dictionary[short_node_name] = short_successors
 

src/_pytask/shared.py

@@ -13,7 +13,6 @@
 from _pytask.node_protocols import MetaNode
 from _pytask.node_protocols import PPathNode
 from _pytask.node_protocols import PTask
-from _pytask.node_protocols import PTaskWithPath
 from _pytask.path import find_closest_ancestor
 from _pytask.path import find_common_ancestor
 from _pytask.path import relative_to
@@ -73,7 +72,7 @@ def reduce_node_name(node: MetaNode, paths: Sequence[Path]) -> str:
     path from one path in ``session.config["paths"]`` to the node.
 
     """
-    if isinstance(node, (PPathNode, PTaskWithPath)):
+    if isinstance(node, PPathNode):
         ancestor = find_closest_ancestor(node.path, paths)
         if ancestor is None:
             try:

tests/test_dag.py

@@ -1,18 +1,14 @@
 from __future__ import annotations
 
 import textwrap
-from contextlib import ExitStack as does_not_raise  # noqa: N813
 from pathlib import Path
 
-import networkx as nx
 import pytest
-from _pytask.dag import _check_if_root_nodes_are_available
 from _pytask.dag import pytask_dag_create_dag
-from _pytask.exceptions import NodeNotFoundError
-from _pytask.exceptions import ResolvingDependenciesError
 from attrs import define
 from pytask import cli
 from pytask import ExitCode
+from pytask import NodeNotFoundError
 from pytask import PathNode
 from pytask import Task
 
@@ -46,36 +42,8 @@ def test_pytask_dag_create_dag():
     )
 
 
-@pytest.mark.unit()
-@pytest.mark.xfail(reason="session object is missing.")
-def test_check_if_root_nodes_are_available():
-    dag = nx.DiGraph()
-
-    root = Path.cwd() / "src"
-
-    path = root.joinpath("task_dummy")
-    task = Task(base_name="task", path=path, function=None)
-    task.path = path
-    task.base_name = "task_dummy"
-    dag.add_node(task.name, task=task)
-
-    available_node = Node.from_path(root.joinpath("available_node"))
-    dag.add_node(available_node.name, node=available_node)
-    dag.add_edge(available_node.name, task.name)
-
-    with does_not_raise():
-        _check_if_root_nodes_are_available(dag)
-
-    missing_node = Node.from_path(root.joinpath("missing_node"))
-    dag.add_node(missing_node.name, node=missing_node)
-    dag.add_edge(missing_node.name, task.name)
-
-    with pytest.raises(ResolvingDependenciesError):
-        _check_if_root_nodes_are_available(dag)
-
-
 @pytest.mark.end_to_end()
-def test_check_if_root_nodes_are_available_end_to_end(tmp_path, runner):
+def test_check_if_root_nodes_are_available(tmp_path, runner):
     source = """
     import pytask
 
@@ -101,9 +69,35 @@ def task_d(produces):
 
 
 @pytest.mark.end_to_end()
-def test_check_if_root_nodes_are_available_with_separate_build_folder_end_to_end(
-    tmp_path, runner
-):
+def test_check_if_root_nodes_are_available_w_name(tmp_path, runner):
+    source = """
+    from pathlib import Path
+    from typing_extensions import Annotated, Any
+    from pytask import PathNode, PythonNode
+
+    node1 = PathNode(name="input1", path=Path(__file__).parent / "in.txt")
+    node2 = PythonNode(name="input2")
+
+    def task_e(in1_: Annotated[Path, node1], in2_: Annotated[Any, node2]): ...
+    """
+    tmp_path.joinpath("task_e.py").write_text(textwrap.dedent(source))
+
+    result = runner.invoke(cli, [tmp_path.as_posix()])
+
+    assert result.exit_code == ExitCode.DAG_FAILED
+    assert "Failures during resolving dependencies" in result.output
+
+    # Ensure that node names are reduced.
+    assert "Failures during resolving dependencies" in result.output
+    assert "Some dependencies do not exist or are" in result.output
+    assert tmp_path.joinpath("task_e.py").as_posix() + "::task_e" not in result.output
+    assert "task_e.py::task_e" in result.output
+    assert tmp_path.joinpath("in.txt").as_posix() not in result.output
+    assert tmp_path.name + "/in.txt" in result.output
+
+
+@pytest.mark.end_to_end()
+def test_check_if_root_nodes_are_available_with_separate_build_folder(tmp_path, runner):
     tmp_path.joinpath("src").mkdir()
     tmp_path.joinpath("bld").mkdir()
     source = """