Implement smooth transition to attributes on nodes.

Author: tobiasraabe

src/_pytask/click.py

@@ -24,6 +24,7 @@
 
 from _pytask import __version__ as version
 from _pytask.console import console
+from _pytask.console import create_panel_title
 
 if TYPE_CHECKING:
     from collections.abc import Sequence
@@ -109,7 +110,7 @@ def format_help(
         console.print(
             Panel(
                 commands_table,
-                title="[bold #f2f2f2]Commands[/]",
+                title=create_panel_title("Commands"),
                 title_align="left",
                 border_style="grey37",
             )
@@ -244,7 +245,7 @@ def _print_options(group_or_command: Command | DefaultGroup, ctx: Context) -> No
     console.print(
         Panel(
             options_table,
-            title="[bold #f2f2f2]Options[/]",
+            title=create_panel_title("Options"),
             title_align="left",
             border_style="grey37",
         )

src/_pytask/collect.py

@@ -7,6 +7,7 @@
 import os
 import sys
 import time
+import warnings
 from contextlib import suppress
 from pathlib import Path
 from typing import TYPE_CHECKING
@@ -385,6 +386,16 @@ def pytask_collect_node(  # noqa: C901, PLR0912
     """
     node = node_info.value
 
+    if isinstance(node, (PNode, PProvisionalNode)) and not hasattr(node, "attributes"):
+        warnings.warn(
+            "PNode and PProvisionalNode will require an 'attributes' field starting "
+            "with pytask v0.6.0. It is a dictionary with any type of key and values "
+            "similar to PTask. See https://tinyurl.com/pytask-custom-nodes for more "
+            "information about adjusting your custom nodes.",
+            stacklevel=1,
+            category=FutureWarning,
+        )
+
     if isinstance(node, DirectoryNode):
         if node.root_dir is None:
             node.root_dir = path

src/_pytask/console.py

@@ -296,10 +296,15 @@ def create_summary_panel(
 
     return Panel(
         grid,
-        title="[bold #f2f2f2]Summary[/]",
+        title=create_panel_title("Summary"),
         expand=False,
         style="none",
         border_style=outcome_enum.FAIL.style
         if counts[outcome_enum.FAIL]
         else outcome_enum.SUCCESS.style,
     )
+
+
+def create_panel_title(title: str) -> Text:
+    """Create a title for a panel."""
+    return Text(title, style="bold #f2f2f2")

src/_pytask/node_protocols.py

@@ -21,7 +21,6 @@ class PNode(Protocol):
     """Protocol for nodes."""
 
     name: str
-    attributes: dict[Any, Any]
 
     @property
     def signature(self) -> str:
@@ -117,7 +116,6 @@ class PProvisionalNode(Protocol):
     """
 
     name: str
-    attributes: dict[Any, Any]
 
     @property
     def signature(self) -> str:

src/_pytask/nodes.py

@@ -162,6 +162,8 @@ class PathNode(PPathNode):
         Name of the node which makes it identifiable in the DAG.
     path
         The path to the file.
+    attributes: dict[Any, Any]
+        A dictionary to store additional information of the task.
 
     """
 
@@ -220,6 +222,8 @@ class PythonNode(PNode):
         objects. The function should return either an integer or a string.
     node_info
         The infos acquired while collecting the node.
+    attributes: dict[Any, Any]
+        A dictionary to store additional information of the task.
 
     Examples
     --------
@@ -304,6 +308,8 @@ class PickleNode(PPathNode):
         Name of the node which makes it identifiable in the DAG.
     path
         The path to the file.
+    attributes: dict[Any, Any]
+        A dictionary to store additional information of the task.
 
     """
 
@@ -353,6 +359,8 @@ class DirectoryNode(PProvisionalNode):
     root_dir
         The pattern is interpreted relative to the path given by ``root_dir``. If
         ``root_dir = None``, it is the directory where the path is defined.
+    attributes: dict[Any, Any]
+        A dictionary to store additional information of the task.
 
     """
 

src/_pytask/warnings.py

@@ -12,6 +12,7 @@
 from rich.panel import Panel
 
 from _pytask.console import console
+from _pytask.console import create_panel_title
 from _pytask.pluginmanager import hookimpl
 from _pytask.warnings_utils import WarningReport
 from _pytask.warnings_utils import catch_warnings_for_item
@@ -82,7 +83,9 @@ def pytask_log_session_footer(session: Session) -> None:
         """Log warnings at the end of a session."""
         if session.warnings:
             renderable = _WarningsRenderable(session.warnings)
-            panel = Panel(renderable, title="Warnings", style="warning")
+            panel = Panel(
+                renderable, title=create_panel_title("Warnings"), style="warning"
+            )
             console.print(panel)
 
 

tests/test_node_protocols.py

@@ -13,6 +13,7 @@
 def test_node_protocol_for_custom_nodes(runner, tmp_path):
     source = """
     from typing import Annotated
+    from typing import Any
     from pytask import Product
     from attrs import define
     from pathlib import Path
@@ -22,6 +23,7 @@ class CustomNode:
         name: str
         value: str
         signature: str = "id"
+        attributes: dict[Any, Any] = {}
 
         def state(self):
             return self.value
@@ -43,12 +45,14 @@ def task_example(
     result = runner.invoke(cli, [tmp_path.as_posix()])
     assert result.exit_code == ExitCode.OK
     assert tmp_path.joinpath("out.txt").read_text() == "text"
+    assert "FutureWarning" not in result.output
 
 
 @pytest.mark.end_to_end
 def test_node_protocol_for_custom_nodes_with_paths(runner, tmp_path):
     source = """
     from typing import Annotated
+    from typing import Any
     from pytask import Product
     from pathlib import Path
     from attrs import define
@@ -60,6 +64,7 @@ class PickleFile:
         path: Path
         value: Path
         signature: str = "id"
+        attributes: dict[Any, Any] = {}
 
         def state(self):
             return str(self.path.stat().st_mtime)
@@ -87,3 +92,40 @@ def task_example(
     result = runner.invoke(cli, [tmp_path.as_posix()])
     assert result.exit_code == ExitCode.OK
     assert tmp_path.joinpath("out.txt").read_text() == "text"
+
+
+@pytest.mark.end_to_end
+def test_node_protocol_for_custom_nodes_adding_attributes(runner, tmp_path):
+    source = """
+    from typing import Annotated
+    from pytask import Product
+    from attrs import define
+    from pathlib import Path
+
+    @define
+    class CustomNode:
+        name: str
+        value: str
+        signature: str = "id"
+
+        def state(self):
+            return self.value
+
+        def load(self, is_product):
+            return self.value
+
+        def save(self, value):
+            self.value = value
+
+    def task_example(
+        data = CustomNode("custom", "text"),
+        out: Annotated[Path, Product] = Path("out.txt"),
+    ) -> None:
+        out.write_text(data)
+    """
+    tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source))
+
+    result = runner.invoke(cli, [tmp_path.as_posix()])
+    assert result.exit_code == ExitCode.OK
+    assert tmp_path.joinpath("out.txt").read_text() == "text"
+    assert "FutureWarning" in result.output