Add a counter under the execution table to show completed tasks.

Author: tobiasraabe

docs/source/changes.md

@@ -34,6 +34,8 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and
 - {pull}`245` adds choices on the command line to the help pages as metavars and show
   defaults.
 - {pull}`246` formalizes choices for {class}`click.Choice` to {class}`enum.Enum`.
+- {pull}`248` adds a counter at the bottom of the execution table to show how many tasks
+  have been processed.
 
 ## 0.1.9 - 2022-02-23
 

src/_pytask/live.py

@@ -5,6 +5,7 @@
 from typing import Dict
 from typing import Generator
 from typing import List
+from typing import Union
 
 import attr
 import click
@@ -16,9 +17,11 @@
 from _pytask.outcomes import TaskOutcome
 from _pytask.report import CollectionReport
 from _pytask.report import ExecutionReport
+from _pytask.session import Session
 from _pytask.shared import get_first_non_none_value
 from rich.live import Live
 from rich.status import Status
+from rich.style import Style
 from rich.table import Table
 from rich.text import Text
 
@@ -83,10 +86,16 @@ def pytask_post_parse(config: dict[str, Any]) -> None:
             config["verbose"],
             config["editor_url_scheme"],
         )
-        config["pm"].register(live_execution)
+        config["pm"].register(live_execution, "live_execution")
 
     live_collection = LiveCollection(live_manager)
-    config["pm"].register(live_collection)
+    config["pm"].register(live_collection, "live_collection")
+
+
+@hookimpl(tryfirst=True)
+def pytask_execute_build(session: Session) -> None:
+    live_execution = session.config["pm"].get_plugin("live_execution")
+    live_execution._n_tasks = len(session.tasks)
 
 
 @attr.s(eq=False)
@@ -148,6 +157,7 @@ class LiveExecution:
     _editor_url_scheme = attr.ib(type=str)
     _running_tasks = attr.ib(factory=dict, type=Dict[str, Task])
     _reports = attr.ib(factory=list, type=List[Dict[str, Any]])
+    _n_tasks = attr.ib(default="x", type=Union[int, str])
 
     @hookimpl(hookwrapper=True)
     def pytask_execute_build(self) -> Generator[None, None, None]:
@@ -156,7 +166,9 @@ def pytask_execute_build(self) -> Generator[None, None, None]:
         self._live_manager.start()
         yield
         self._live_manager.stop(transient=True)
-        table = self._generate_table(reduce_table=False, sort_table=True)
+        table = self._generate_table(
+            reduce_table=False, sort_table=True, add_caption=False
+        )
         if table is not None:
             console.print(table)
 
@@ -172,7 +184,9 @@ def pytask_execute_task_log_end(self, report: ExecutionReport) -> bool:
         self.update_reports(report)
         return True
 
-    def _generate_table(self, reduce_table: bool, sort_table: bool) -> Table | None:
+    def _generate_table(
+        self, reduce_table: bool, sort_table: bool, add_caption: bool
+    ) -> Table | None:
         """Generate the table.
 
         First, display all completed tasks and, then, all running tasks.
@@ -210,7 +224,19 @@ def _generate_table(self, reduce_table: bool, sort_table: bool) -> Table | None:
                 relevant_reports, key=lambda report: report["name"]
             )
 
-        table = Table()
+        if add_caption:
+            caption_kwargs = {
+                "caption": Text(
+                    f"Completed: {len(self._reports)}/{self._n_tasks}",
+                    style=Style(dim=True, italic=False),
+                ),
+                "caption_justify": "right",
+                "caption_style": None,
+            }
+        else:
+            caption_kwargs = {}
+
+        table = Table(**caption_kwargs)
         table.add_column("Task", overflow="fold")
         table.add_column("Outcome")
         for report in relevant_reports:
@@ -237,10 +263,15 @@ def _generate_table(self, reduce_table: bool, sort_table: bool) -> Table | None:
         return table
 
     def _update_table(
-        self, reduce_table: bool = True, sort_table: bool = False
+        self,
+        reduce_table: bool = True,
+        sort_table: bool = False,
+        add_caption: bool = True,
     ) -> None:
         """Regenerate the table."""
-        table = self._generate_table(reduce_table=reduce_table, sort_table=sort_table)
+        table = self._generate_table(
+            reduce_table=reduce_table, sort_table=sort_table, add_caption=add_caption
+        )
         self._live_manager.update(table)
 
     def update_running_tasks(self, new_running_task: Task) -> None:

tests/test_live.py

@@ -77,6 +77,7 @@ def test_live_execution_sequentially(capsys, tmp_path):
     assert "Outcome" not in captured.out
     assert "task_module.py::task_example" not in captured.out
     assert "running" not in captured.out
+    assert "Completed: 0/x" not in captured.out
 
     live_manager.resume()
     live_manager.start()
@@ -88,6 +89,7 @@ def test_live_execution_sequentially(capsys, tmp_path):
     assert "Outcome" in captured.out
     assert "task_module.py::task_example" in captured.out
     assert "running" in captured.out
+    assert "Completed: 0/x" in captured.out
 
     live_manager.start()
 
@@ -104,6 +106,7 @@ def test_live_execution_sequentially(capsys, tmp_path):
     assert "task_module.py::task_example" in captured.out
     assert "running" not in captured.out
     assert TaskOutcome.SUCCESS.symbol in captured.out
+    assert "Completed: 1/x" in captured.out
 
 
 @pytest.mark.unit
@@ -160,6 +163,7 @@ def test_live_execution_displays_subset_of_table(capsys, tmp_path, n_entries_in_
 
     live_manager = LiveManager()
     live = LiveExecution(live_manager, n_entries_in_table, 1, "no_link")
+    live._n_tasks = 2
 
     live_manager.start()
     live.update_running_tasks(running_task)
@@ -170,6 +174,7 @@ def test_live_execution_displays_subset_of_table(capsys, tmp_path, n_entries_in_
     assert "Outcome" in captured.out
     assert "::task_running" in captured.out
     assert " running " in captured.out
+    assert "Completed: 0/2" in captured.out
 
     completed_task = Task(base_name="task_completed", path=path, function=lambda x: x)
     completed_task.short_name = "task_module.py::task_completed"
@@ -188,6 +193,7 @@ def test_live_execution_displays_subset_of_table(capsys, tmp_path, n_entries_in_
     assert "Outcome" in captured.out
     assert "::task_running" in captured.out
     assert " running " in captured.out
+    assert "Completed: 1/2" in captured.out
 
     if n_entries_in_table == 1:
         assert "task_module.py::task_completed" not in captured.out