Refactor remaining pieces related to reduce_node_name. (#184)

Author: tobiasraabe

src/_pytask/collect.py

@@ -17,6 +17,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_task_id
 from _pytask.exceptions import CollectionError
 from _pytask.mark_utils import has_marker
 from _pytask.nodes import create_task_name
@@ -111,7 +112,7 @@ def pytask_collect_file(
         spec = importlib_util.spec_from_file_location(path.stem, str(path))
 
         if spec is None:
-            raise ImportError(f"Can't find module '{path.stem}' at location {path}.")
+            raise ImportError(f"Can't find module {path.stem!r} at location {path}.")
 
         mod = importlib_util.module_from_spec(spec)
         spec.loader.exec_module(mod)
@@ -340,7 +341,12 @@ def pytask_collect_log(
             if report.node is None:
                 header = "Error"
             else:
-                short_name = reduce_node_name(report.node, session.config["paths"])
+                if isinstance(report.node, MetaTask):
+                    short_name = format_task_id(
+                        report.node, editor_url_scheme="no_link", short_name=True
+                    )
+                else:
+                    short_name = reduce_node_name(report.node, session.config["paths"])
                 header = f"Could not collect {short_name}"
 
             console.rule(

src/_pytask/compat.py

@@ -77,8 +77,8 @@ def import_optional_dependency(
     if extra and not extra.endswith(" "):
         extra += " "
     msg = (
-        f"{caller} requires the optional dependency '{install_name}'. {extra}"
-        f"Use pip or conda to install '{install_name}'."
+        f"{caller} requires the optional dependency {install_name!r}. {extra}"
+        f"Use pip or conda to install {install_name!r}."
     )
     try:
         module = importlib.import_module(name)
@@ -102,8 +102,8 @@ def import_optional_dependency(
         version = _get_version(module_to_get)
         if parse_version(version) < parse_version(minimum_version):
             msg = (
-                f"{caller} requires version '{minimum_version}' or newer of "
-                f"'{parent}' (version '{version}' currently installed)."
+                f"{caller} requires version {minimum_version!r} or newer of "
+                f"{parent!r} (version {version!r} currently installed)."
             )
             if errors == "warn":
                 warnings.warn(msg, UserWarning)
@@ -123,10 +123,10 @@ def check_for_optional_program(
     """Check whether an optional program exists."""
     if errors not in ("warn", "raise", "ignore"):
         raise ValueError(
-            f"'errors' must be one of 'warn', 'raise' or 'ignore' and not '{errors}'."
+            f"'errors' must be one of 'warn', 'raise' or 'ignore' and not {errors!r}."
         )
 
-    msg = f"{caller} requires the optional program '{name}'. {extra}"
+    msg = f"{caller} requires the optional program {name!r}. {extra}"
 
     program_exists = shutil.which(name) is not None
 

src/_pytask/console.py

@@ -88,7 +88,12 @@
 console = Console(theme=theme, color_system=_COLOR_SYSTEM)
 
 
-def render_to_string(text: Union[str, Text], console: Optional[Console] = None) -> str:
+def render_to_string(
+    text: Union[str, Text],
+    *,
+    console: Optional[Console] = None,
+    strip_styles: bool = False,
+) -> str:
     """Render text with rich to string including ANSI codes, etc..
 
     This function allows to render text with is not automatically printed with rich. For
@@ -104,6 +109,9 @@ def render_to_string(text: Union[str, Text], console: Optional[Console] = None)
     if console.no_color and console._color_system:
         segments = Segment.remove_color(segments)
 
+    if strip_styles:
+        segments = Segment.strip_styles(segments)
+
     for segment in segments:
         if segment.style:
             output.append(
@@ -134,7 +142,12 @@ def format_task_id(
         task_name = task.base_name
     else:
         path, task_name = task.name.split("::")
-    url_style = create_url_style_for_task(task, editor_url_scheme)
+
+    if task.function is None:
+        url_style = Style()
+    else:
+        url_style = create_url_style_for_task(task.function, editor_url_scheme)
+
     task_id = Text.assemble(
         Text(path + "::", style="dim"), Text(task_name, style=url_style)
     )
@@ -146,17 +159,19 @@ def format_strings_as_flat_tree(strings: Iterable[str], title: str, icon: str) -
     tree = Tree(title)
     for name in strings:
         tree.add(icon + name)
-    text = render_to_string(tree, console)
+    text = render_to_string(tree, console=console)
     return text
 
 
-def create_url_style_for_task(task: "MetaTask", edtior_url_scheme: str) -> Style:
+def create_url_style_for_task(
+    task_function: Callable[..., Any], edtior_url_scheme: str
+) -> Style:
     """Create the style to add a link to a task id."""
     url_scheme = _EDITOR_URL_SCHEMES.get(edtior_url_scheme, edtior_url_scheme)
 
     info = {
-        "path": _get_file(task.function),
-        "line_number": _get_source_lines(task.function),
+        "path": _get_file(task_function),
+        "line_number": _get_source_lines(task_function),
     }
 
     return Style() if not url_scheme else Style(link=url_scheme.format(**info))

src/_pytask/execute.py

@@ -9,6 +9,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_task_id
 from _pytask.console import unify_styles
 from _pytask.dag import descending_tasks
 from _pytask.dag import TopologicalSorter
@@ -24,7 +25,6 @@
 from _pytask.report import ExecutionReport
 from _pytask.session import Session
 from _pytask.shared import get_first_non_none_value
-from _pytask.shared import reduce_node_name
 from _pytask.traceback import format_exception_without_traceback
 from _pytask.traceback import remove_traceback_from_exc_info
 from _pytask.traceback import render_exc_info
@@ -186,7 +186,7 @@ def pytask_execute_task_process_report(
                 Mark(
                     "skip_ancestor_failed",
                     (),
-                    {"reason": f"Previous task '{task.name}' failed."},
+                    {"reason": f"Previous task {task.name!r} failed."},
                 )
             )
 
@@ -204,7 +204,7 @@ def pytask_execute_task_process_report(
 def pytask_execute_task_log_end(session: Session, report: ExecutionReport) -> None:
     """Log task outcome."""
     url_style = create_url_style_for_task(
-        report.task, session.config["editor_url_scheme"]
+        report.task.function, session.config["editor_url_scheme"]
     )
     console.print(
         report.outcome.symbol,
@@ -260,21 +260,13 @@ def pytask_execute_log_end(session: Session, reports: List[ExecutionReport]) ->
 
 def _print_errored_task_report(session: Session, report: ExecutionReport) -> None:
     """Print the traceback and the exception of an errored report."""
-    task_name = reduce_node_name(report.task, session.config["paths"])
-    if len(task_name) > console.width - 15:
-        task_name = report.task.base_name
-
-    url_style = create_url_style_for_task(
-        report.task, session.config["editor_url_scheme"]
-    )
-
-    console.rule(
-        Text(
-            f"Task {task_name} failed",
-            style=unify_styles(report.outcome.style, url_style),
-        ),
-        style=report.outcome.style,
+    task_name = format_task_id(
+        task=report.task,
+        editor_url_scheme=session.config["editor_url_scheme"],
+        short_name=True,
     )
+    text = Text.assemble("Task ", task_name, " failed", style="failed")
+    console.rule(text, style=report.outcome.style)
 
     console.print()
 

src/_pytask/graph.py

@@ -22,6 +22,7 @@
 from _pytask.shared import get_first_non_none_value
 from _pytask.shared import reduce_names_of_multiple_nodes
 from _pytask.traceback import remove_internal_traceback_frames_from_exc_info
+from rich.text import Text
 from rich.traceback import Traceback
 
 
@@ -232,6 +233,7 @@ def _shorten_node_labels(dag: nx.DiGraph, paths: List[Path]) -> nx.DiGraph:
     """Shorten the node labels in the graph for a better experience."""
     node_names = dag.nodes
     short_names = reduce_names_of_multiple_nodes(node_names, dag, paths)
+    short_names = [i.plain if isinstance(i, Text) else i for i in short_names]
     old_to_new = dict(zip(node_names, short_names))
     dag = nx.relabel_nodes(dag, old_to_new)
     return dag

src/_pytask/logging.py

@@ -207,7 +207,7 @@ def _humanize_time(
             index = i
             break
     else:
-        raise ValueError(f"The time unit '{unit}' is not known.")
+        raise ValueError(f"The time unit {unit!r} is not known.")
 
     seconds = amount * _TIME_UNITS[index]["in_seconds"]
 

src/_pytask/mark/structures.py

@@ -224,7 +224,7 @@ def __getattr__(self, name: str) -> MarkDecorator:
                     raise ValueError(f"Unknown pytask.mark.{name}.")
                 # Raise a specific error for common misspellings of "parametrize".
                 if name in ["parameterize", "parametrise", "parameterise"]:
-                    warnings.warn(f"Unknown '{name}' mark, did you mean 'parametrize'?")
+                    warnings.warn(f"Unknown {name!r} mark, did you mean 'parametrize'?")
 
                 warnings.warn(
                     f"Unknown pytask.mark.{name} - is this a typo? You can register "

src/_pytask/nodes.py

@@ -260,8 +260,8 @@ def _collect_nodes(
         )
         if collected_node is None:
             raise NodeNotCollectedError(
-                f"'{node}' cannot be parsed as a dependency or product for task "
-                f"'{name}' in '{path}'."
+                f"{node!r} cannot be parsed as a dependency or product for task "
+                f"{name!r} in {path!r}."
             )
         else:
             collected_nodes[node_name] = collected_node

src/_pytask/parametrize.py

@@ -139,7 +139,7 @@ def pytask_parametrize_task(
             )
             raise ValueError(
                 "The following ids are duplicated while parametrizing task "
-                f"'{name}'.\n\n{text}\n\nIt might be caused by "
+                f"{name!r}.\n\n{text}\n\nIt might be caused by "
                 "parametrizing the task with the same combination of arguments "
                 "multiple times. Change the arguments or change the ids generated by "
                 "the parametrization."
@@ -296,7 +296,7 @@ def _check_if_n_arg_names_matches_n_arg_values(
         idx_example = [i == n_names for i in n_values].index(False)
         formatted_example = pprint.pformat(arg_values[idx_example])
         raise ValueError(
-            f"Task '{name}' is parametrized with {n_names} 'arg_names', {arg_names}, "
+            f"Task {name!r} is parametrized with {n_names} 'arg_names', {arg_names}, "
             f"but the number of provided 'arg_values' is {pretty_arg_values}. For "
             f"example, here are the values of parametrization no. {idx_example}:"
             f"\n\n{formatted_example}"

src/_pytask/profile.py

@@ -272,7 +272,7 @@ def pytask_profile_export_profile(
         elif extension is None:
             pass
         else:
-            raise ValueError(f"The export option '{extension}' cannot be handled.")
+            raise ValueError(f"The export option {extension!r} cannot be handled.")
 
 
 def _export_to_csv(profile: Dict[str, Dict[str, Any]]) -> None:

src/_pytask/resolve_dependencies.py

@@ -265,7 +265,7 @@ def _format_dictionary_to_tree(dict_: Dict[str, List[str]], title: str) -> str:
         for task in tasks:
             branch.add(Text.assemble(TASK_ICON, task))
 
-    text = render_to_string(tree)
+    text = render_to_string(tree, console=console, strip_styles=True)
     return text
 
 

src/_pytask/shared.py

@@ -10,7 +10,7 @@
 from typing import Union
 
 import networkx as nx
-from _pytask.nodes import create_task_name
+from _pytask.console import format_task_id
 from _pytask.nodes import MetaNode
 from _pytask.nodes import MetaTask
 from _pytask.path import find_closest_ancestor
@@ -120,7 +120,7 @@ def parse_value_or_multiline_option(
     elif isinstance(value, str):
         return value.strip()
     else:
-        raise ValueError(f"Input '{value}' is neither a 'str' nor 'None'.")
+        raise ValueError(f"Input {value!r} is neither a 'str' nor 'None'.")
 
 
 def convert_truthy_or_falsy_to_bool(x: Union[bool, str, None]) -> bool:
@@ -133,7 +133,7 @@ def convert_truthy_or_falsy_to_bool(x: Union[bool, str, None]) -> bool:
         out = None
     else:
         raise ValueError(
-            f"Input '{x}' is neither truthy (True, true, 1) or falsy (False, false, 0)."
+            f"Input {x!r} is neither truthy (True, true, 1) or falsy (False, false, 0)."
         )
     return out
 
@@ -155,13 +155,10 @@ def reduce_node_name(node: "MetaNode", paths: Sequence[Union[str, Path]]) -> str
         except ValueError:
             ancestor = node.path.parents[-1]
 
-    if isinstance(node, MetaTask):
-        shortened_path = relative_to(node.path, ancestor)
-        name = create_task_name(shortened_path, node.base_name)
-    elif isinstance(node, MetaNode):
+    if isinstance(node, MetaNode):
         name = relative_to(node.path, ancestor).as_posix()
     else:
-        raise TypeError(f"Unknown node {node} with type '{type(node)}'.")
+        raise TypeError(f"Unknown node {node} with type {type(node)!r}.")
 
     return name
 
@@ -170,7 +167,21 @@ def reduce_names_of_multiple_nodes(
     names: List[str], dag: nx.DiGraph, paths: Sequence[Union[str, Path]]
 ) -> List[str]:
     """Reduce the names of multiple nodes in the DAG."""
-    return [
-        reduce_node_name(dag.nodes[n].get("node") or dag.nodes[n].get("task"), paths)
-        for n in names
-    ]
+    short_names = []
+    for name in names:
+        node = dag.nodes[name].get("node") or dag.nodes[name].get("task")
+
+        if isinstance(node, MetaTask):
+            short_name = format_task_id(
+                node, editor_url_scheme="no_link", short_name=True
+            )
+        elif isinstance(node, MetaNode):
+            short_name = reduce_node_name(node, paths)
+        else:
+            raise TypeError(
+                f"Requires 'MetaTask' or 'MetaNode' and not {type(node)!r}."
+            )
+
+        short_names.append(short_name)
+
+    return short_names

src/_pytask/skipping.py

@@ -97,7 +97,7 @@ def pytask_execute_task_process_report(
                     Mark(
                         "skip",
                         (),
-                        {"reason": f"Previous task '{task.name}' was skipped."},
+                        {"reason": f"Previous task {task.name!r} was skipped."},
                     )
                 )