Pretty format tasks with mixed priorities, simplify find_common_ancestor function and others. (#80)

Author: tobiasraabe

docs/_static/css/custom.css

@@ -1,8 +1,3 @@
-/* Remove execution count for notebook cells. */
-div.prompt {
-    display: none;
-}
-
 img.card-img-top {
     height: 52px;
 }

docs/changes.rst

@@ -6,6 +6,13 @@ chronological order. Releases follow `semantic versioning <https://semver.org/>`
 all releases are available on `PyPI <https://pypi.org/project/pytask>`_ and
 `Anaconda.org <https://anaconda.org/conda-forge/pytask>`_.
 
+
+0.0.15 - 2021-xx-xx
+-------------------
+
+- :gh:`80` replaces some remaining formatting using ``pprint`` with ``rich``.
+
+
 0.0.14 - 2021-03-23
 -------------------
 

docs/conf.py

@@ -7,18 +7,22 @@
 # root, use os.path.abspath to make it absolute, like shown here.
 import os
 import sys
+from datetime import datetime
 
-import pytask
 import sphinx
 
 
 sys.path.insert(0, os.path.abspath("../src"))
 
 
+import pytask  # noqa: E402
+
+
 # -- Project information ---------------------------------------------------------------
 
 project = "pytask"
-copyright = "2020, Tobias Raabe"  # noqa: A001
+year = datetime.now().year
+copyright = f"2020-{year}, Tobias Raabe"  # noqa: A001
 author = "Tobias Raabe"
 
 # The full version, including alpha/beta/rc tags

docs/rtd_environment.yml

@@ -24,6 +24,3 @@ dependencies:
   - pony >=0.7.13
   - pexpect
   - rich
-
-  - pip:
-    - -e ../

src/_pytask/console.py

@@ -1,9 +1,10 @@
 """This module contains the code to format output on the command line."""
 import os
 import sys
+from typing import List
 
 from rich.console import Console
-
+from rich.tree import Tree
 
 _IS_WSL = "IS_WSL" in os.environ or "WSL_DISTRO_NAME" in os.environ
 _IS_WINDOWS_TERMINAL = "WT_SESSION" in os.environ
@@ -26,3 +27,16 @@
 
 
 console = Console(color_system=_COLOR_SYSTEM)
+
+
+def format_strings_as_flat_tree(strings: List[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]
+    )
+
+    return text

src/_pytask/dag.py

@@ -1,15 +1,18 @@
 """Implement some capabilities to deal with the DAG."""
 import itertools
-import pprint
+from pathlib import Path
 from typing import Dict
 from typing import Generator
 from typing import Iterable
 from typing import List
 
 import attr
 import networkx as nx
+from _pytask.console import format_strings_as_flat_tree
+from _pytask.console import TASK_ICON
 from _pytask.mark import get_specific_markers_from_task
 from _pytask.nodes import MetaTask
+from _pytask.nodes import reduce_node_name
 
 
 def descending_tasks(task_name: str, dag: nx.DiGraph) -> Generator[str, None, None]:
@@ -52,14 +55,17 @@ class TopologicalSorter:
     _nodes_out = attr.ib(factory=set)
 
     @classmethod
-    def from_dag(cls, dag: nx.DiGraph) -> "TopologicalSorter":
+    def from_dag(cls, dag: nx.DiGraph, paths: List[Path] = None) -> "TopologicalSorter":
+        if paths is None:
+            paths = []
+
         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)
+        priorities = _extract_priorities_from_tasks(tasks, paths)
 
         task_names = {task.name for task in tasks}
         task_dict = {name: nx.ancestors(dag, name) & task_names for name in task_names}
@@ -118,7 +124,9 @@ def static_order(self) -> Generator[str, None, None]:
             self.done(new_task)
 
 
-def _extract_priorities_from_tasks(tasks: List[MetaTask]) -> Dict[str, int]:
+def _extract_priorities_from_tasks(
+    tasks: List[MetaTask], paths: List[Path]
+) -> Dict[str, int]:
     """Extract priorities from tasks.
 
     Priorities are set via the ``pytask.mark.try_first`` and ``pytask.mark.try_last``
@@ -138,10 +146,19 @@ def _extract_priorities_from_tasks(tasks: List[MetaTask]) -> Dict[str, int]:
     tasks_w_mixed_priorities = [
         name for name, p in priorities.items() if p["try_first"] and p["try_last"]
     ]
+
     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
+        ]
+        text = format_strings_as_flat_tree(
+            reduced_names, "Tasks with mixed priorities", TASK_ICON
+        )
         raise ValueError(
             "'try_first' and 'try_last' cannot be applied on the same task. See the "
-            f"following tasks for errors:\n\n{pprint.pformat(tasks_w_mixed_priorities)}"
+            f"following tasks for errors:\n\n{text}"
         )
 
     # Recode to numeric values for sorting.

src/_pytask/nodes.py

@@ -366,3 +366,11 @@ def reduce_node_name(node, paths: List[Path]):
         raise ValueError(f"Unknown node {node} with type '{type(node)}'.")
 
     return name
+
+
+def reduce_names_of_multiple_nodes(names, dag, paths):
+    """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
+    ]

src/_pytask/parametrize.py

@@ -1,7 +1,6 @@
 import copy
 import functools
 import itertools
-import pprint
 import types
 from typing import Any
 from typing import Callable
@@ -12,7 +11,8 @@
 from typing import Union
 
 from _pytask.config import hookimpl
-from _pytask.console import console
+from _pytask.console import format_strings_as_flat_tree
+from _pytask.console import TASK_ICON
 from _pytask.mark import Mark
 from _pytask.mark import MARK_GEN as mark  # noqa: N811
 from _pytask.shared import find_duplicates
@@ -124,10 +124,12 @@ def pytask_parametrize_task(session, name, obj):
         names = [i[0] for i in names_and_functions]
         duplicates = find_duplicates(names)
         if duplicates:
-            formatted = pprint.pformat(duplicates, width=console.width)
+            text = format_strings_as_flat_tree(
+                duplicates, "Duplicated task ids", TASK_ICON
+            )
             raise ValueError(
                 "The following ids are duplicated while parametrizing task "
-                f"{obj.__name__}.\n\n{formatted}\n\nIt might be caused by "
+                f"'{obj.__name__}'.\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."

src/_pytask/path.py

@@ -1,6 +1,6 @@
 """This module contains code to handle paths."""
+import os
 from pathlib import Path
-from pathlib import PurePath
 from typing import List
 from typing import Union
 
@@ -83,19 +83,6 @@ def find_common_ancestor_of_nodes(*names: str) -> Path:
 
 def find_common_ancestor(*paths: Union[str, Path]) -> Path:
     """Find a common ancestor of many paths."""
-    paths = [path if isinstance(path, PurePath) else Path(path) for path in paths]
-
-    for path in paths:
-        if not path.is_absolute():
-            raise ValueError(
-                f"Cannot find common ancestor for relative paths. {path} is relative."
-            )
-
-    common_parents = set.intersection(*[set(path.parents) for path in paths])
-
-    if len(common_parents) == 0:
-        raise ValueError("Paths have no common ancestor.")
-    else:
-        longest_parent = sorted(common_parents, key=lambda x: len(x.parts))[-1]
-
-    return longest_parent
+    path = os.path.commonpath(paths)
+    path = Path(path)
+    return path

src/_pytask/resolve_dependencies.py

@@ -18,6 +18,7 @@
 from _pytask.exceptions import NodeNotFoundError
 from _pytask.exceptions import ResolvingDependenciesError
 from _pytask.mark import Mark
+from _pytask.nodes import reduce_names_of_multiple_nodes
 from _pytask.nodes import reduce_node_name
 from _pytask.path import find_common_ancestor_of_nodes
 from _pytask.report import ResolvingDependenciesReport
@@ -180,7 +181,7 @@ def _check_if_root_nodes_are_available(dag):
             short_node_name = reduce_node_name(
                 dag.nodes[node]["node"], [common_ancestor]
             )
-            short_successors = _reduce_names_of_multiple_nodes(
+            short_successors = reduce_names_of_multiple_nodes(
                 dag.successors(node), dag, [common_ancestor]
             )
             dictionary[short_node_name] = short_successors
@@ -231,7 +232,7 @@ def _check_if_tasks_have_the_same_products(dag):
             short_node_name = reduce_node_name(
                 dag.nodes[node]["node"], [common_ancestor]
             )
-            short_predecessors = _reduce_names_of_multiple_nodes(
+            short_predecessors = reduce_names_of_multiple_nodes(
                 dag.predecessors(node), dag, [common_ancestor]
             )
             dictionary[short_node_name] = short_predecessors
@@ -257,11 +258,3 @@ def pytask_resolve_dependencies_log(report):
 
     console.print()
     console.rule(style=ColorCode.FAILED)
-
-
-def _reduce_names_of_multiple_nodes(names, dag, paths):
-    """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
-    ]