Shorten and colorize task ids. (#178)

Author: tobiasraabe

.pre-commit-config.yaml

@@ -30,7 +30,7 @@ repos:
     -   id: rst-inline-touching-normal
     -   id: text-unicode-replacement-char
 -   repo: https://github.com/asottile/pyupgrade
-    rev: v2.29.1
+    rev: v2.31.0
     hooks:
     -   id: pyupgrade
         args: [--py36-plus]
@@ -90,10 +90,9 @@ repos:
     hooks:
     -   id: check-manifest
 -   repo: https://github.com/guilatrova/tryceratops
-    rev: v1.0.0
+    rev: v1.0.1
     hooks:
     -   id: tryceratops
-        exclude: (console\.py|test_mark_expression\.py)
 -   repo: https://github.com/pre-commit/mirrors-mypy
     rev: 'v0.930'
     hooks:

MANIFEST.in

@@ -5,6 +5,5 @@ exclude *.yaml
 exclude *.yml
 exclude tox.ini
 
-prune .conda
 prune docs
 prune tests

docs/source/changes.rst

@@ -22,6 +22,8 @@ all releases are available on `PyPI <https://pypi.org/project/pytask>`_ and
 - :gh:`174` restructures loosely defined outcomes to clear ``enum.Enum``.
 - :gh:`176` and :gh:`177` implement a summary panel which holds aggregate information
   about the number of successes, fails and other status.
+- :gh:`178` makes some stylistic changes like reducing tasks ids even more and dims the
+  path part.
 
 
 0.1.3 - 2021-11-30

docs/source/tutorials/how_to_collect_tasks.rst

@@ -8,7 +8,7 @@ For example, let us take the following task
 
 .. code-block:: python
 
-    # Content of task_dummy.py
+    # Content of task_module.py
 
     import pytask
 
@@ -29,7 +29,7 @@ Now, running :program:`pytask collect` will produce the following output.
     Root: xxx
     Collected 1 task(s).
 
-    <Module /.../task_dummy.py>
+    <Module /.../task_module.py>
       <Function task_write_file>
 
     ========================================================================
@@ -45,7 +45,7 @@ append the ``--nodes`` flag.
     Root: xxx
     Collected 1 task(s).
 
-    <Module /.../task_dummy.py>
+    <Module /.../task_module.py>
       <Function task_write_file>
         <Dependency /.../in.txt>
         <Product /.../out.txt>

src/_pytask/collect.py

@@ -21,6 +21,8 @@
 from _pytask.mark_utils import has_marker
 from _pytask.nodes import create_task_name
 from _pytask.nodes import FilePathNode
+from _pytask.nodes import find_duplicates
+from _pytask.nodes import MetaTask
 from _pytask.nodes import PythonFunctionTask
 from _pytask.outcomes import CollectionOutcome
 from _pytask.outcomes import count_outcomes
@@ -82,6 +84,7 @@ def pytask_ignore_collect(path: Path, config: Dict[str, Any]) -> bool:
 def pytask_collect_file_protocol(
     session: Session, path: Path, reports: List[CollectionReport]
 ) -> List[CollectionReport]:
+    """Wrap the collection of tasks from a file to collect reports."""
     try:
         reports = session.hook.pytask_collect_file(
             session=session, path=path, reports=reports
@@ -268,6 +271,52 @@ def _not_ignored_paths(
                 yield path
 
 
+@hookimpl(trylast=True)
+def pytask_collect_modify_tasks(tasks: List[MetaTask]) -> None:
+    """Given all tasks, assign a short uniquely identifiable name to each task.
+
+    The shorter ids are necessary to display
+
+    """
+    id_to_short_id = _find_shortest_uniquely_identifiable_name_for_tasks(tasks)
+    for task in tasks:
+        short_id = id_to_short_id[task.name]
+        task.short_name = short_id
+
+
+def _find_shortest_uniquely_identifiable_name_for_tasks(
+    tasks: List[MetaTask],
+) -> Dict[str, str]:
+    """Find the shortest uniquely identifiable name for tasks.
+
+    The shortest unique id consists of the module name plus the base name (e.g. function
+    name) of the task. If this does not make the id unique, append more and more parent
+    folders until the id is unique.
+
+    """
+    id_to_short_id = {}
+
+    # Make attempt to add up to twenty parts of the path to ensure uniqueness.
+    id_to_task = {task.name: task for task in tasks}
+    for n_parts in range(1, 20):
+        dupl_id_to_short_id = {
+            id_: "/".join(task.path.parts[-n_parts:]) + "::" + task.base_name
+            for id_, task in id_to_task.items()
+        }
+        duplicates = find_duplicates(dupl_id_to_short_id.values())
+
+        for id_, short_id in dupl_id_to_short_id.items():
+            if short_id not in duplicates:
+                id_to_short_id[id_] = short_id
+                id_to_task.pop(id_)
+
+    # If there are still non-unique task ids, just use the full id as the short id.
+    for id_, task in id_to_task.items():
+        id_to_short_id[id_] = task.name
+
+    return id_to_short_id
+
+
 @hookimpl
 def pytask_collect_log(
     session: Session, reports: List[CollectionReport], tasks: List[PythonFunctionTask]

src/_pytask/collect_command.py

@@ -11,8 +11,8 @@
 from _pytask.config import hookimpl
 from _pytask.console import console
 from _pytask.console import create_url_style_for_path
-from _pytask.console import create_url_style_for_task
 from _pytask.console import FILE_ICON
+from _pytask.console import format_task_id
 from _pytask.console import PYTHON_ICON
 from _pytask.console import TASK_ICON
 from _pytask.enums import ExitCode
@@ -25,7 +25,6 @@
 from _pytask.path import relative_to
 from _pytask.pluginmanager import get_plugin_manager
 from _pytask.session import Session
-from _pytask.shared import reduce_node_name
 from rich.text import Text
 from rich.tree import Tree
 
@@ -194,15 +193,11 @@ def _print_collected_tasks(
         )
 
         for task in tasks:
-            reduced_task_name = reduce_node_name(task, [common_ancestor])
-            url_style = create_url_style_for_task(task, editor_url_scheme)
+            reduced_task_name = format_task_id(
+                task, editor_url_scheme=editor_url_scheme, relative_to=common_ancestor
+            )
             task_branch = module_branch.add(
-                Text.assemble(
-                    TASK_ICON,
-                    "<Function ",
-                    Text(reduced_task_name, style=url_style),
-                    ">",
-                ),
+                Text.assemble(TASK_ICON, "<Function ", reduced_task_name, ">"),
             )
 
             if show_nodes:

src/_pytask/console.py

@@ -9,15 +9,20 @@
 from typing import Callable
 from typing import Dict
 from typing import Iterable
+from typing import Optional
 from typing import Type
 from typing import TYPE_CHECKING
 from typing import Union
 
+import rich
+from _pytask.path import relative_to as relative_to_
 from rich.console import Console
 from rich.padding import Padding
 from rich.panel import Panel
+from rich.segment import Segment
 from rich.style import Style
 from rich.table import Table
+from rich.text import Text
 from rich.theme import Theme
 from rich.tree import Tree
 
@@ -76,43 +81,75 @@
 console = Console(theme=theme, color_system=_COLOR_SYSTEM)
 
 
+def render_to_string(text: Union[str, Text], console: Optional[Console] = None) -> 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
+    example, render warnings with colors or text in exceptions.
+
+    """
+    if console is None:
+        console = rich.get_console()
+
+    segments = console.render(text)
+
+    output = []
+    if console.no_color and console._color_system:
+        segments = Segment.remove_color(segments)
+
+    for segment in segments:
+        if segment.style:
+            output.append(
+                segment.style.render(
+                    segment.text,
+                    color_system=console._color_system,
+                    legacy_windows=console.legacy_windows,
+                )
+            )
+        else:
+            output.append(segment.text)
+
+    rendered = "".join(output)
+    return rendered
+
+
+def format_task_id(
+    task: "MetaTask",
+    editor_url_scheme: str,
+    short_name: bool = False,
+    relative_to: Optional[Path] = None,
+) -> Text:
+    """Format a task id."""
+    if short_name:
+        path, task_name = task.short_name.split("::")
+    elif relative_to:
+        path = relative_to_(task.path, relative_to).as_posix()
+        task_name = task.base_name
+    else:
+        path, task_name = task.name.split("::")
+    url_style = create_url_style_for_task(task, editor_url_scheme)
+    task_id = Text.assemble(
+        Text(path + "::", style="dim"), Text(task_name, style=url_style)
+    )
+    return task_id
+
+
 def format_strings_as_flat_tree(strings: Iterable[str], title: str, icon: str) -> str:
     """Format list of strings as flat tree."""
     tree = Tree(title)
     for name in strings:
         tree.add(icon + name)
-
-    text = "".join(
-        [x.text for x in tree.__rich_console__(console, console.options)][:-1]
-    )
-
+    text = render_to_string(tree, console)
     return text
 
 
-def escape_squared_brackets(string: str) -> str:
-    """Escape squared brackets which would be accidentally parsed by rich.
-
-    An example are the ids of parametrized tasks which are suffixed with squared
-    brackets surrounding string representations of the parametrized arguments.
-
-    Example
-    -------
-    >>> escape_squared_brackets("Hello!")
-    'Hello!'
-    >>> escape_squared_brackets("task_dummy[arg1-arg2]")
-    'task_dummy\\\\[arg1-arg2]'
-
-    """
-    return string.replace("[", "\\[")
-
-
 def create_url_style_for_task(task: "MetaTask", 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": inspect.getsourcelines(task.function)[1],
+        "line_number": _get_source_lines(task.function),
     }
 
     return Style() if not url_scheme else Style(link=url_scheme.format(**info))
@@ -136,6 +173,14 @@ def _get_file(function: Callable[..., Any]) -> Path:
         return Path(inspect.getfile(function))
 
 
+def _get_source_lines(function: Callable[..., Any]) -> int:
+    """Get the source line number of the function."""
+    if isinstance(function, functools.partial):
+        return _get_source_lines(function.func)
+    else:
+        return inspect.getsourcelines(function)[1]
+
+
 def unify_styles(*styles: Union[str, Style]) -> Style:
     """Unify styles."""
     parsed_styles = []

src/_pytask/dag.py

@@ -1,6 +1,5 @@
 """Implement some capabilities to deal with the DAG."""
 import itertools
-from pathlib import Path
 from typing import Dict
 from typing import Generator
 from typing import Iterable
@@ -13,7 +12,6 @@
 from _pytask.console import TASK_ICON
 from _pytask.mark_utils import get_specific_markers_from_task
 from _pytask.nodes import MetaTask
-from _pytask.shared import reduce_node_name
 
 
 def descending_tasks(task_name: str, dag: nx.DiGraph) -> Generator[str, None, None]:
@@ -71,17 +69,15 @@ class TopologicalSorter:
     _nodes_out = attr.ib(factory=set, type=Set[str])
 
     @classmethod
-    def from_dag(cls, dag: nx.DiGraph, paths: List[Path] = None) -> "TopologicalSorter":
-        if paths is None:
-            paths = []
-
+    def from_dag(cls, dag: nx.DiGraph) -> "TopologicalSorter":
+        """Instantiate from a DAG."""
         if not dag.is_directed():
             raise ValueError("Only directed graphs have a topological order.")
 
         tasks = [
             dag.nodes[node]["task"] for node in dag.nodes if "task" in dag.nodes[node]
         ]
-        priorities = _extract_priorities_from_tasks(tasks, paths)
+        priorities = _extract_priorities_from_tasks(tasks)
 
         task_names = {task.name for task in tasks}
         task_dict = {name: nx.ancestors(dag, name) & task_names for name in task_names}
@@ -140,9 +136,7 @@ def static_order(self) -> Generator[str, None, None]:
             self.done(new_task)
 
 
-def _extract_priorities_from_tasks(
-    tasks: List[MetaTask], paths: List[Path]
-) -> Dict[str, int]:
+def _extract_priorities_from_tasks(tasks: List[MetaTask]) -> Dict[str, int]:
     """Extract priorities from tasks.
 
     Priorities are set via the ``pytask.mark.try_first`` and ``pytask.mark.try_last``
@@ -166,8 +160,7 @@ def _extract_priorities_from_tasks(
     if tasks_w_mixed_priorities:
         name_to_task = {task.name: task for task in tasks}
         reduced_names = [
-            reduce_node_name(name_to_task[name], paths)
-            for name in tasks_w_mixed_priorities
+            name_to_task[name].short_name for name in tasks_w_mixed_priorities
         ]
         text = format_strings_as_flat_tree(
             reduced_names, "Tasks with mixed priorities", TASK_ICON

src/_pytask/database.py

@@ -134,6 +134,7 @@ def pytask_parse_config(
 
 @hookimpl
 def pytask_post_parse(config: Dict[str, Any]) -> None:
+    """Post-parse the configuration."""
     create_database(**config["database"])
 
 

src/_pytask/debugging.py

@@ -455,6 +455,7 @@ def wrapper(*args: Any, **kwargs: Any) -> None:
 
 
 def post_mortem(t: TracebackType) -> None:
+    """Start post-mortem debugging."""
     p = PytaskPDB._init_pdb("post_mortem")
     p.reset()
     p.interaction(None, t)