VSCode Logging

src/_pytask/pluginmanager.py

@@ -58,6 +58,7 @@ def pytask_add_hooks(pm: PluginManager) -> None:
         "_pytask.skipping",
         "_pytask.task",
         "_pytask.warnings",
+        "_pytask.vscode",
     )
     register_hook_impls_from_modules(pm, builtin_hook_impl_modules)
 

src/_pytask/vscode.py

@@ -0,0 +1,96 @@
+"""Contains Code for VSCode Logging."""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import os
+from threading import Thread
+from typing import TYPE_CHECKING
+from typing import Any
+from urllib import request
+
+from _pytask.config import hookimpl
+from _pytask.console import console
+from _pytask.console import render_to_string
+from _pytask.nodes import PTaskWithPath
+from _pytask.outcomes import CollectionOutcome
+from _pytask.outcomes import TaskOutcome
+from _pytask.traceback import Traceback
+
+if TYPE_CHECKING:
+    from _pytask.node_protocols import PTask
+    from _pytask.reports import CollectionReport
+    from _pytask.reports import ExecutionReport
+    from _pytask.session import Session
+
+
+def send_logging_vscode(url: str, data: dict[str, Any], timeout: float) -> None:
+    """Send logging information to VSCode."""
+    with contextlib.suppress(Exception):
+        response = json.dumps(data).encode("utf-8")
+        req = request.Request(url, data=response)  # noqa: S310
+        req.add_header("Content-Type", "application/json; charset=utf-8")
+        request.urlopen(req, timeout=timeout)  # noqa: S310
+
+
+@hookimpl(tryfirst=True)
+def pytask_collect_log(
+    session: Session, reports: list[CollectionReport], tasks: list[PTask]
+) -> None:
+    if (
+        os.environ.get("PYTASK_VSCODE") is not None
+        and session.config["command"] == "collect"
+    ):
+        try:
+            port = int(os.environ.get("PYTASK_VSCODE"))
+        except ValueError:
+            port = 6000
+        exitcode = "OK"
+        for report in reports:
+            if report.outcome == CollectionOutcome.FAIL:
+                exitcode = "COLLECTION_FAILED"
+        result = [
+            {"name": task.name, "path": str(task.path)}
+            if isinstance(task, PTaskWithPath)
+            else {"name": task.name, "path": ""}
+            for task in tasks
+        ]
+        url = f"http://localhost:{port}/pytask/collect"
+        thread = Thread(
+            target=send_logging_vscode,
+            args=(
+                url,
+                {"exitcode": exitcode, "tasks": result},
+                0.00001,
+            ),
+        )
+        thread.start()
+
+
+@hookimpl(tryfirst=True)
+def pytask_execute_task_log_end(session: Session, report: ExecutionReport) -> None:  # noqa: ARG001
+    if os.environ.get("PYTASK_VSCODE") is not None:
+        try:
+            port = int(os.environ.get("PYTASK_VSCODE"))
+        except ValueError:
+            port = 6000
+        if report.outcome == TaskOutcome.FAIL and report.exc_info is not None:
+            result = {
+                "type": "task",
+                "name": report.task.name,
+                "outcome": str(report.outcome),
+                "exc_info": render_to_string(Traceback(report.exc_info), console),
+            }
+        else:
+            result = {
+                "type": "task",
+                "name": report.task.name,
+                "outcome": str(report.outcome),
+            }
+        url = f"http://localhost:{port}/pytask/run"
+        thread = Thread(
+            target=send_logging_vscode,
+            args=(url, result, 0.00001),
+        )
+        thread.start()

tests/test_vscode.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import os
+import textwrap
+
+import pytest
+from pytask import ExitCode
+from pytask import cli
+
+
+@pytest.mark.end_to_end()
+def test_vscode_collect_failed(runner, tmp_path):
+    source = """
+    raise Exception
+    """
+    os.environ["PYTASK_VSCODE"] = "6000"
+    tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source))
+
+    result = runner.invoke(cli, ["collect", tmp_path.as_posix()])
+    assert result.exit_code == ExitCode.COLLECTION_FAILED
+
+
+@pytest.mark.end_to_end()
+def test_vscode_collect(runner, tmp_path):
+    source = """
+    def task_raises():
+        return
+    """
+    os.environ["PYTASK_VSCODE"] = "6000"
+    tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source))
+
+    result = runner.invoke(cli, ["collect", tmp_path.as_posix()])
+    assert result.exit_code == ExitCode.OK
+
+
+@pytest.mark.end_to_end()
+def test_vscode_build(runner, tmp_path):
+    source = """
+    def task_example():
+        return
+    def task_raises():
+        raise Exception
+    """
+    os.environ["PYTASK_VSCODE"] = "6000"
+    tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source))
+
+    result = runner.invoke(cli, [tmp_path.as_posix()])
+    assert result.exit_code == ExitCode.FAILED