Fix display issues with the programmatic interface. (#631)

Author: tobiasraabe

docs/source/changes.md

@@ -18,6 +18,8 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and
   module's name does not start with `task_`.
 - {pull}`628` fixes duplicated collection of task modules. Fixes {issue}`624`. Thanks to
   {user}`timmens` for the issue.
+- {pull}`631` fixes display issues with the programmatic interface by giving each
+  {class}`~_pytask.live.LiveManager` its own {class}`~rich.live.Live`.
 
 ## 0.5.0 - 2024-05-26
 

docs/source/tutorials/visualizing_the_dag.md

@@ -35,11 +35,13 @@ layouts, which are listed [here](https://graphviz.org/docs/layouts/).
 
 ## Programmatic Interface
 
-The programmatic and interactive interface allows customizing the figure.
+The programmatic and interactive interface allows for customizing the figure.
 
 Similar to {func}`pytask.build`, there exists {func}`pytask.build_dag` which returns the
 DAG as a {class}`networkx.DiGraph`.
 
+Create an executable script that you can execute with `python script.py`.
+
 ```{literalinclude} ../../../docs_src/tutorials/visualizing_the_dag.py
 ```
 

docs_src/tutorials/visualizing_the_dag.py

@@ -3,12 +3,10 @@
 import networkx as nx
 from my_project.config import BLD
 from my_project.config import SRC
-from pytask import Product
 from pytask import build_dag
-from typing_extensions import Annotated
 
 
-def task_draw_dag(path: Annotated[Path, Product] = BLD / "dag.svg") -> None:
+def draw_dag(path: Path = BLD / "dag.svg") -> None:
     dag = build_dag({"paths": SRC})
 
     # Set shapes to hexagons.
@@ -17,3 +15,7 @@ def task_draw_dag(path: Annotated[Path, Product] = BLD / "dag.svg") -> None:
     # Export with pygraphviz and dot.
     graph = nx.nx_agraph.to_agraph(dag)
     graph.draw(path, prog="dot")
+
+
+if __name__ == "__main__":
+    draw_dag()

pyproject.toml

@@ -90,7 +90,9 @@ build-backend = "hatchling.build"
 
 [tool.rye]
 managed = true
-dev-dependencies = ["tox-uv>=1.7.0"]
+dev-dependencies = [
+    "tox-uv>=1.7.0",
+]
 
 [tool.rye.scripts]
 clean-docs = { cmd = "rm -rf docs/build" }

src/_pytask/live.py

@@ -12,6 +12,7 @@
 from attrs import define
 from attrs import field
 from rich.box import ROUNDED
+from rich.errors import LiveError
 from rich.live import Live
 from rich.status import Status
 from rich.style import Style
@@ -22,6 +23,7 @@
 from _pytask.console import format_task_name
 from _pytask.logging_utils import TaskExecutionStatus
 from _pytask.outcomes import CollectionOutcome
+from _pytask.outcomes import Exit
 from _pytask.outcomes import TaskOutcome
 from _pytask.pluginmanager import hookimpl
 
@@ -102,10 +104,21 @@ class LiveManager:
 
     """
 
-    _live = Live(renderable=None, console=console, auto_refresh=False)
+    _live: Live = field(
+        factory=lambda: Live(renderable=None, console=console, auto_refresh=False)
+    )
 
     def start(self) -> None:
-        self._live.start()
+        try:
+            self._live.start()
+        except LiveError:
+            msg = (
+                "pytask tried to launch a second live display which is impossible. the "
+                "issue occurs when you use pytask on the command line on a task module "
+                "that uses the programmatic interface of pytask at the same time. "
+                "Use either the command line or the programmatic interface."
+            )
+            raise Exit(msg) from None
 
     def stop(self, transient: bool | None = None) -> None:
         if transient is not None:
@@ -319,8 +332,6 @@ def pytask_collect_log(self) -> Generator[None, None, None]:
 
     def _update_statistics(self, reports: list[CollectionReport]) -> None:
         """Update the statistics on collected tasks and errors."""
-        if reports is None:
-            reports = []
         for report in reports:
             if report.outcome == CollectionOutcome.SUCCESS:
                 self._n_collected_tasks += 1

tests/test_dag_command.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+import subprocess
 import sys
 import textwrap
 
@@ -65,7 +66,7 @@ def task_example(path=Path("input.txt")): ...
 @pytest.mark.parametrize("layout", _GRAPH_LAYOUTS)
 @pytest.mark.parametrize("format_", _TEST_FORMATS)
 @pytest.mark.parametrize("rankdir", [_RankDirection.LR.value, _RankDirection.TB])
-def test_create_graph_via_task(tmp_path, runner, format_, layout, rankdir):
+def test_create_graph_via_task(tmp_path, format_, layout, rankdir):
     if sys.platform == "win32" and format_ == "pdf":  # pragma: no cover
         pytest.xfail("gvplugin_pango.dll might be missing on Github Actions.")
 
@@ -78,20 +79,24 @@ def test_create_graph_via_task(tmp_path, runner, format_, layout, rankdir):
 
     def task_example(path=Path("input.txt")): ...
 
-    def task_create_graph():
+    def main():
         dag = pytask.build_dag({{"paths": Path(__file__).parent}})
         dag.graph = {{"rankdir": "{rankdir_str}"}}
         graph = nx.nx_agraph.to_agraph(dag)
         path = Path(__file__).parent.joinpath("dag.{format_}")
         graph.draw(path, prog="{layout}")
-    """
 
+    if __name__ == "__main__":
+        main()
+    """
     tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(source))
     tmp_path.joinpath("input.txt").touch()
 
-    result = runner.invoke(cli, [tmp_path.as_posix()])
+    result = subprocess.run(
+        ("python", "task_example.py"), cwd=tmp_path, check=True, capture_output=True
+    )
 
-    assert result.exit_code == ExitCode.OK
+    assert result.returncode == ExitCode.OK
     assert tmp_path.joinpath(f"dag.{format_}").exists()
 
 

tests/test_execute.py

@@ -625,6 +625,24 @@ def create_file(
     assert tmp_path.joinpath("file.txt").read_text() == "This is the text."
 
 
+@pytest.mark.end_to_end()
+def test_pytask_on_a_module_that_uses_the_functional_api(tmp_path):
+    source = """
+    from pytask import task, ExitCode, build
+    from pathlib import Path
+    from typing_extensions import Annotated
+
+    def task_example(): pass
+
+    session = build(tasks=[task_example])
+    assert session.exit_code == ExitCode.OK
+    """
+    tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(source))
+    result = subprocess.run(("pytask",), cwd=tmp_path, capture_output=True, check=False)
+    assert result.returncode == ExitCode.COLLECTION_FAILED
+    assert "pytask tried to launch a second live display" in result.stdout.decode()
+
+
 @pytest.mark.end_to_end()
 def test_pass_non_task_to_functional_api_that_are_ignored():
     session = pytask.build(tasks=None)

tests/test_task.py

@@ -688,7 +688,7 @@ def task_second():
     """
     tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(source))
 
-    result = subprocess.run(("pytask",), cwd=tmp_path, capture_output=True, check=False)
+    result = subprocess.run(("python",), cwd=tmp_path, capture_output=True, check=False)
     assert result.returncode == ExitCode.OK