Cli

docs/source/tutorial/3-troubleshooting.ipynb

@@ -141,7 +141,7 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "The error pickle files can be loaded using the `cloudpickle` library, noting that it is\n",
+    "The error pickle files can be viewed using the `pydracli crash` command, with the possibility of rerunning and debugging the job. Note that it is\n",
     "important to use the same Python version to load the files that was used to run the Pydra\n",
     "workflow"
    ]
@@ -152,17 +152,28 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "from pydra.utils.general import default_run_cache_root\n",
-    "import cloudpickle as cp\n",
-    "from pprint import pprint\n",
     "from pydra.tasks.testing import Divide\n",
+    "from pydra.utils.general import default_run_cache_root\n",
     "\n",
-    "with open(\n",
-    "    default_run_cache_root / Divide(x=15, y=0)._checksum / \"_error.pklz\", \"rb\"\n",
-    ") as f:\n",
-    "    error = cp.load(f)\n",
+    "if __name__ == \"__main__\":\n",
+    "    divide = Divide(x=15, y=0)\n",
+    "    try:\n",
+    "        divide(cache_root=default_run_cache_root)\n",
+    "    except Exception:\n",
+    "        pass\n",
+    "\n",
+    "    errorfile = default_run_cache_root / divide._checksum / \"_error.pklz\""
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "%%bash -s \"$errorfile\"\n",
     "\n",
-    "pprint(error)"
+    "pydracli crash $1"
    ]
   },
   {

pydra/conftest.py

@@ -1,6 +1,8 @@
 import shutil
 import os
 import pytest
+import typing as ty
+from click.testing import CliRunner, Result as CliResult
 
 os.environ["NO_ET"] = "true"
 
@@ -41,17 +43,39 @@
         metafunc.parametrize("any_worker", available_workers)
 
 
+@pytest.fixture
+def cli_runner(catch_cli_exceptions: bool) -> ty.Callable[..., ty.Any]:
+    def invoke(
+        *args: ty.Any, catch_exceptions: bool = catch_cli_exceptions, **kwargs: ty.Any
+    ) -> CliResult:
+        runner = CliRunner()
+        result = runner.invoke(*args, catch_exceptions=catch_exceptions, **kwargs)  # type: ignore[misc]
+        return result
+
+    return invoke
+
+
 # For debugging in IDE's don't catch raised exceptions and let the IDE
 # break at it
-if os.getenv("_PYTEST_RAISE", "0") != "0":  # pragma: no cover
+if os.getenv("_PYTEST_RAISE", "0") != "0":
+
+    @pytest.hookimpl(tryfirst=True)
+    def pytest_exception_interact(call: pytest.CallInfo[ty.Any]) -> None:
+        if call.excinfo is not None:
+            raise call.excinfo.value
+
+    @pytest.hookimpl(tryfirst=True)
+    def pytest_internalerror(excinfo: pytest.ExceptionInfo[BaseException]) -> None:
+        raise excinfo.value
+
+    CATCH_CLI_EXCEPTIONS = False
+else:
+    CATCH_CLI_EXCEPTIONS = True
 
-    @pytest.hookimpl(tryfirst=True)  # pragma: no cover
-    def pytest_exception_interact(call):  # pragma: no cover
-        raise call.excinfo.value  # pragma: no cover
 
-    @pytest.hookimpl(tryfirst=True)  # pragma: no cover
-    def pytest_internalerror(excinfo):  # pragma: no cover
-        raise excinfo.value  # pragma: no cover
+@pytest.fixture
+def catch_cli_exceptions() -> bool:
+    return CATCH_CLI_EXCEPTIONS
 
 
 # Example VSCode launch configuration for debugging unittests

pydra/scripts/cli.py

@@ -0,0 +1,63 @@
+from pathlib import Path
+import pdb
+import sys
+
+import click
+import cloudpickle as cp
+
+CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
+ExistingFilePath = click.Path(exists=True, dir_okay=False, resolve_path=True)
+
+
+@click.group(context_settings=CONTEXT_SETTINGS)
+def cli():
+    pass
+
+
+@cli.command(context_settings=CONTEXT_SETTINGS)
+@click.argument("crashfile", type=ExistingFilePath)
+@click.option(
+    "-r", "--rerun", is_flag=True, flag_value=True, help="Rerun crashed code."
+)
+@click.option(
+    "-d",
+    "--debugger",
+    type=click.Choice([None, "ipython", "pdb"]),
+    help="Debugger to use when rerunning",
+)
+def crash(crashfile, rerun, debugger=None):
+    """Display a crash file and rerun if required."""
+    if crashfile.endswith(("pkl", "pklz")):
+        with open(crashfile, "rb") as f:
+            crash_content = cp.load(f)
+        print("".join(crash_content["error message"]))
+
+        if rerun:
+            jobfile = Path(crashfile).parent / "_job.pklz"
+            if jobfile.exists():
+                with open(jobfile, "rb") as f:
+                    job_obj = cp.load(f)
+
+                if debugger == "ipython":
+                    try:
+                        from IPython.core import ultratb
+
+                        sys.excepthook = ultratb.FormattedTB(
+                            mode="Verbose", theme_name="Linux", call_pdb=True
+                        )
+                    except ImportError:
+                        raise ImportError(
+                            "'Ipython' needs to be installed to use the 'ipython' debugger"
+                        )
+
+                try:
+                    job_obj.run(rerun=True)
+                except Exception:  # noqa: E722
+                    if debugger == "pdb":
+                        pdb.post_mortem()
+                    elif debugger == "ipython":
+                        raise
+            else:
+                raise FileNotFoundError(f"Job file {jobfile} not found")
+    else:
+        raise ValueError("Only pickled crashfiles are supported")

pydra/scripts/tests/test_crash.py

@@ -0,0 +1,28 @@
+import pytest
+from pydra.scripts.cli import crash
+from pydra.tasks.testing import Divide
+from traceback import format_exception
+import typing as ty
+
+
+# @pytest.mark.xfail(reason="Need to fix a couple of things after syntax changes")
+def test_crash_cli(cli_runner, tmp_path):
+    divide = Divide(x=15, y=0)
+    with pytest.raises(ZeroDivisionError):
+        divide(cache_root=tmp_path)
+
+    result = cli_runner(
+        crash,
+        [
+            f"{tmp_path}/{divide._checksum}/_error.pklz",
+            "--rerun",
+            "--debugger",
+            "pdb",
+        ],
+    )
+    assert result.exit_code == 0, show_cli_trace(result)
+
+
+def show_cli_trace(result: ty.Any) -> str:
+    "Used in testing to show traceback of CLI output"
+    return "".join(format_exception(*result.exc_info))

pyproject.toml

@@ -114,6 +114,9 @@ documentation = "https://nipype.github.io/pydra/"
 homepage = "https://nipype.github.io/pydra/"
 repository = "https://github.com/nipype/pydra.git"
 
+[project.scripts]
+pydracli = "pydra.scripts.cli:cli"
+
 [tool.hatch.build]
 packages = ["pydra"]
 exclude = ["tests"]