Refactor code for format_node_name.

docs/source/changes.md

@@ -13,6 +13,7 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and
 - {pull}`455` adds more explanation when {meth}`~pytask.PNode.load` fails during the
   execution.
 - {pull}`456` refers to the source code on Github when clicking on a source link.
+- {pull}`457` refactors everything around formatting node names.
 
 ## 0.4.1 - 2023-10-11
 

src/_pytask/collect.py

@@ -19,6 +19,7 @@
 from _pytask.config import IS_FILE_SYSTEM_CASE_SENSITIVE
 from _pytask.console import console
 from _pytask.console import create_summary_panel
+from _pytask.console import format_node_name
 from _pytask.console import format_task_name
 from _pytask.console import get_file
 from _pytask.console import is_jupyter
@@ -38,7 +39,6 @@
 from _pytask.path import import_path
 from _pytask.report import CollectionReport
 from _pytask.shared import find_duplicates
-from _pytask.shared import reduce_node_name
 from _pytask.task_utils import task as task_decorator
 from _pytask.traceback import render_exc_info
 from _pytask.typing import is_task_function
@@ -460,8 +460,17 @@ def pytask_collect_log(
                     short_name = format_task_name(
                         report.node, editor_url_scheme="no_link"
                     ).plain
+                elif isinstance(report.node, PNode):
+                    short_name = format_node_name(
+                        report.node, session.config["paths"]
+                    ).plain
                 else:
-                    short_name = reduce_node_name(report.node, session.config["paths"])
+                    msg = (
+                        "Requires a 'PTask' or a 'PNode' and not "
+                        f"{type(report.node)!r}."
+                    )
+                    raise TypeError(msg)
+
                 header = f"Could not collect {short_name}"
 
             console.rule(

src/_pytask/collect_command.py

@@ -12,6 +12,7 @@
 from _pytask.console import console
 from _pytask.console import create_url_style_for_path
 from _pytask.console import FILE_ICON
+from _pytask.console import format_node_name
 from _pytask.console import format_task_name
 from _pytask.console import PYTHON_ICON
 from _pytask.console import TASK_ICON
@@ -202,14 +203,7 @@ def _print_collected_tasks(
                         else x.name
                     ),
                 ):
-                    if isinstance(node, PPathNode):
-                        reduced_node_name = str(relative_to(node.path, common_ancestor))
-                        url_style = create_url_style_for_path(
-                            node.path, editor_url_scheme
-                        )
-                        text = Text(reduced_node_name, style=url_style)
-                    else:
-                        text = Text(node.name)
+                    text = format_node_name(node, (common_ancestor,))
                     task_branch.add(Text.assemble(FILE_ICON, "<Dependency ", text, ">"))
 
                 products: list[PNode] = list(tree_leaves(task.produces))  # type: ignore[arg-type]
@@ -219,14 +213,7 @@ def _print_collected_tasks(
                     if isinstance(x, PPathNode)
                     else x.name,
                 ):
-                    if isinstance(node, PPathNode):
-                        reduced_node_name = str(relative_to(node.path, common_ancestor))
-                        url_style = create_url_style_for_path(
-                            node.path, editor_url_scheme
-                        )
-                        text = Text(reduced_node_name, style=url_style)
-                    else:
-                        text = Text(node.name)
+                    text = format_node_name(node, (common_ancestor,))
                     task_branch.add(Text.assemble(FILE_ICON, "<Product ", text, ">"))
 
     console.print(tree)

src/_pytask/console.py

@@ -5,15 +5,20 @@
 import inspect
 import os
 import sys
+from contextlib import suppress
 from pathlib import Path
 from typing import Any
 from typing import Callable
 from typing import Iterable
 from typing import Literal
+from typing import Sequence
 from typing import TYPE_CHECKING
 
 import rich
-from _pytask.nodes import Task
+from _pytask.node_protocols import PNode
+from _pytask.node_protocols import PPathNode
+from _pytask.node_protocols import PTaskWithPath
+from _pytask.path import shorten_path
 from rich.console import Console
 from rich.console import RenderableType
 from rich.padding import Padding
@@ -149,15 +154,31 @@ def format_task_name(task: PTask, editor_url_scheme: str) -> Text:
     else:
         url_style = create_url_style_for_task(task.function, editor_url_scheme)
 
-    if isinstance(task, Task):
+    if isinstance(task, PTaskWithPath) and hasattr(task, "display_name"):
         path, task_name = task.display_name.split("::")
-        task_id = Text.assemble(
+        return Text.assemble(
             Text(path + "::", style="dim"), Text(task_name, style=url_style)
         )
-    else:
-        name = getattr(task, "display_name", task.name)
-        task_id = Text(name, style=url_style)
-    return task_id
+
+    name = getattr(task, "display_name", task.name)
+    return Text(name, style=url_style)
+
+
+def format_node_name(node: PNode, paths: Sequence[Path] = ()) -> Text:
+    """Format the name of a node."""
+    if isinstance(node, PPathNode):
+        if node.name != node.path.as_posix():
+            return Text(node.name)
+        name = shorten_path(node.path, paths)
+        return Text(name)
+
+    if "::" in node.name:
+        with suppress(Exception):
+            path, rest = node.name.split("::", maxsplit=1)
+            reduced_name = shorten_path(Path(path), paths)
+            return Text(f"{reduced_name}::{rest}")
+
+    return Text(node.name)
 
 
 def format_strings_as_flat_tree(strings: Iterable[str], title: str, icon: str) -> str:

src/_pytask/dag.py

@@ -13,6 +13,7 @@
 from _pytask.console import ARROW_DOWN_ICON
 from _pytask.console import console
 from _pytask.console import FILE_ICON
+from _pytask.console import format_node_name
 from _pytask.console import render_to_string
 from _pytask.console import TASK_ICON
 from _pytask.dag_utils import node_and_neighbors
@@ -32,7 +33,6 @@
 from _pytask.nodes import PythonNode
 from _pytask.report import DagReport
 from _pytask.shared import reduce_names_of_multiple_nodes
-from _pytask.shared import reduce_node_name
 from _pytask.traceback import remove_internal_traceback_frames_from_exception
 from _pytask.traceback import render_exc_info
 from _pytask.tree_util import tree_map
@@ -251,7 +251,7 @@ def _check_if_root_nodes_are_available(dag: nx.DiGraph, paths: Sequence[Path]) -
     if missing_root_nodes:
         dictionary = {}
         for node in missing_root_nodes:
-            short_node_name = reduce_node_name(dag.nodes[node]["node"], paths)
+            short_node_name = format_node_name(dag.nodes[node]["node"], paths).plain
             not_skipped_successors = [
                 task for task in dag.successors(node) if not is_task_skipped[task]
             ]
@@ -329,7 +329,7 @@ def _check_if_tasks_have_the_same_products(dag: nx.DiGraph, paths: list[Path]) -
     if nodes_created_by_multiple_tasks:
         dictionary = {}
         for node in nodes_created_by_multiple_tasks:
-            short_node_name = reduce_node_name(dag.nodes[node]["node"], paths)
+            short_node_name = format_node_name(dag.nodes[node]["node"], paths).plain
             short_predecessors = reduce_names_of_multiple_nodes(
                 dag.predecessors(node), dag, paths
             )

src/_pytask/execute.py

@@ -11,6 +11,7 @@
 from _pytask.console import console
 from _pytask.console import create_summary_panel
 from _pytask.console import create_url_style_for_task
+from _pytask.console import format_node_name
 from _pytask.console import format_strings_as_flat_tree
 from _pytask.console import format_task_name
 from _pytask.console import unify_styles
@@ -31,7 +32,6 @@
 from _pytask.outcomes import TaskOutcome
 from _pytask.outcomes import WouldBeExecuted
 from _pytask.report import ExecutionReport
-from _pytask.shared import reduce_node_name
 from _pytask.traceback import format_exception_without_traceback
 from _pytask.traceback import remove_internal_traceback_frames_from_exception
 from _pytask.traceback import remove_traceback_from_exc_info
@@ -203,7 +203,9 @@ def pytask_execute_task_teardown(session: Session, task: PTask) -> None:
             missing_nodes.append(node)
 
     if missing_nodes:
-        paths = [reduce_node_name(i, session.config["paths"]) for i in missing_nodes]
+        paths = [
+            format_node_name(i, session.config["paths"]).plain for i in missing_nodes
+        ]
         formatted = format_strings_as_flat_tree(
             paths, "The task did not produce the following files:\n", ""
         )

src/_pytask/path.py

@@ -11,6 +11,16 @@
 from typing import Sequence
 
 
+__all__ = [
+    "find_case_sensitive_path",
+    "find_closest_ancestor",
+    "find_common_ancestor",
+    "import_path",
+    "relative_to",
+    "shorten_path",
+]
+
+
 def relative_to(path: Path, source: Path, include_source: bool = True) -> Path:
     """Make a path relative to another path.
 
@@ -176,3 +186,23 @@ def _insert_missing_modules(modules: dict[str, ModuleType], module_name: str) ->
                 modules[module_name] = module
         module_parts.pop(-1)
         module_name = ".".join(module_parts)
+
+
+def shorten_path(path: Path, paths: Sequence[Path]) -> str:
+    """Shorten a path.
+
+    The whole path of a node - which includes the drive letter - can be very long
+    when using nested folder structures in bigger projects.
+
+    Thus, the part of the name which contains the path is replaced by the relative
+    path from one path in ``session.config["paths"]`` to the node.
+
+    """
+    ancestor = find_closest_ancestor(path, paths)
+    if ancestor is None:
+        try:
+            ancestor = find_common_ancestor(path, *paths)
+        except ValueError:
+            ancestor = path.parents[-1]
+
+    return relative_to(path, ancestor).as_posix()

src/_pytask/shared.py

@@ -9,13 +9,10 @@
 from typing import TYPE_CHECKING
 
 import click
+from _pytask.console import format_node_name
 from _pytask.console import format_task_name
-from _pytask.node_protocols import MetaNode
-from _pytask.node_protocols import PPathNode
+from _pytask.node_protocols import PNode
 from _pytask.node_protocols import PTask
-from _pytask.path import find_closest_ancestor
-from _pytask.path import find_common_ancestor
-from _pytask.path import relative_to
 
 if TYPE_CHECKING:
     import networkx as nx
@@ -62,28 +59,6 @@ def parse_paths(x: Any | None) -> list[Path] | None:
     return out
 
 
-def reduce_node_name(node: MetaNode, paths: Sequence[Path]) -> str:
-    """Reduce the node name.
-
-    The whole name of the node - which includes the drive letter - can be very long
-    when using nested folder structures in bigger projects.
-
-    Thus, the part of the name which contains the path is replaced by the relative
-    path from one path in ``session.config["paths"]`` to the node.
-
-    """
-    if isinstance(node, PPathNode):
-        ancestor = find_closest_ancestor(node.path, paths)
-        if ancestor is None:
-            try:
-                ancestor = find_common_ancestor(node.path, *paths)
-            except ValueError:
-                ancestor = node.path.parents[-1]
-
-        return relative_to(node.path, ancestor).as_posix()
-    return node.name
-
-
 def reduce_names_of_multiple_nodes(
     names: list[str], dag: nx.DiGraph, paths: Sequence[Path]
 ) -> list[str]:
@@ -94,10 +69,10 @@ def reduce_names_of_multiple_nodes(
 
         if isinstance(node, PTask):
             short_name = format_task_name(node, editor_url_scheme="no_link").plain
-        elif isinstance(node, MetaNode):
-            short_name = reduce_node_name(node, paths)
+        elif isinstance(node, PNode):
+            short_name = format_node_name(node, paths).plain
         else:
-            msg = f"Requires 'Task' or 'Node' and not {type(node)!r}."
+            msg = f"Requires a 'PTask' or a 'PNode' and not {type(node)!r}."
             raise TypeError(msg)
 
         short_names.append(short_name)

tests/test_collect_command.py

@@ -628,7 +628,8 @@ def task_example() -> Annotated[Dict[str, str], nodes]:
     tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source))
     result = runner.invoke(cli, ["collect", "--nodes", tmp_path.as_posix()])
     assert result.exit_code == ExitCode.OK
-    assert "return::0" in result.output
-    assert "return::1-0" in result.output
-    assert "return::1-1" in result.output
-    assert "return::2" in result.output
+    output = result.output.replace(" ", "").replace("\n", "").replace("│", "")
+    assert "return::0" in output
+    assert "return::1-0" in output
+    assert "return::1-1" in output
+    assert "return::2" in output

tests/test_console.py

@@ -1,18 +1,22 @@
 from __future__ import annotations
 
 import inspect
+from contextlib import ExitStack as does_not_raise  # noqa: N813
 from pathlib import Path
 
 import pytest
 from _pytask.console import _get_source_lines
 from _pytask.console import create_summary_panel
 from _pytask.console import create_url_style_for_path
 from _pytask.console import create_url_style_for_task
+from _pytask.console import format_node_name
 from _pytask.console import format_task_name
 from _pytask.console import get_file
 from _pytask.console import render_to_string
 from pytask import CollectionOutcome
 from pytask import console
+from pytask import PathNode
+from pytask import PythonNode
 from pytask import Task
 from pytask import TaskOutcome
 from rich.console import Console
@@ -145,6 +149,49 @@ def test_format_task_id(
     assert result == expected
 
 
+_ROOT = Path.cwd()
+
+
+@pytest.mark.integration()
+@pytest.mark.parametrize(
+    ("node", "paths", "expectation", "expected"),
+    [
+        pytest.param(
+            PathNode.from_path(_ROOT.joinpath("src/module.py")),
+            [_ROOT.joinpath("alternative_src")],
+            does_not_raise(),
+            Text("pytask/src/module.py"),
+            id="Common path found for PathNode not in 'paths' and 'paths'",
+        ),
+        pytest.param(
+            PathNode.from_path(_ROOT.joinpath("top/src/module.py")),
+            [_ROOT.joinpath("top/src")],
+            does_not_raise(),
+            Text("src/module.py"),
+            id="make filepathnode relative to 'paths'.",
+        ),
+        pytest.param(
+            PythonNode(name="hello", value=None),
+            [_ROOT],
+            does_not_raise(),
+            Text("hello"),
+            id="PythonNode with name",
+        ),
+        pytest.param(
+            PythonNode(name=_ROOT.as_posix() + "/task_a.py::task_a::a", value=None),
+            [_ROOT],
+            does_not_raise(),
+            Text("pytask/task_a.py::task_a::a"),
+            id="PythonNode with automatically assigned name",
+        ),
+    ],
+)
+def test_reduce_node_name(node, paths, expectation, expected):
+    with expectation:
+        result = format_node_name(node, paths)
+        assert result == expected
+
+
 @pytest.mark.unit()
 @pytest.mark.parametrize(
     ("task_func", "skipped_paths", "expected"),

tests/test_dag.py

@@ -94,7 +94,7 @@ def task_e(in1_: Annotated[Path, node1], in2_: Annotated[Any, node2]): ...
     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
+    assert "input1" in result.output
 
 
 @pytest.mark.end_to_end()