pytask-vscode/src/pytask_vscode/execution.py at 9fd7a30eae168cfac24c3ccae67631f92b4de08e · pytask-dev/pytask-vscode · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
from __future__ import annotations


import pytask
import requests


@pytask.hookimpl(tryfirst=True)
def pytask_collect_log(
    session: pytask.Session,
    reports: list[pytask.CollectionReport],
    tasks: list[pytask.PTask],
) -> None:
    try:
        if session.config["command"] == "collect":
            exitcode = 0
            for report in reports:
                if report.outcome == pytask.CollectionOutcome.FAIL:
                    exitcode = 3
            result = [
                (
                    {"name": task.name.split("/")[-1], "path": str(task.path)}
                    if isinstance(task, pytask.PTaskWithPath)
                    else {"name": task.name, "path": ""}
                )
                for task in tasks
            ]
            res = requests.post(
                "http://localhost:6000/pytask",
                json={"exitcode": exitcode, "tasks": result},
                timeout=0.0001,
            )
    except requests.exceptions.ReadTimeout:
        pass
    except Exception:
        pass


@pytask.hookimpl(tryfirst=True)
def pytask_execute_task_log_end(
    session: pytask.Session, report: pytask.ExecutionReport
) -> None:

    try:
        if report.outcome == pytask.TaskOutcome.FAIL:
            with pytask.console.capture() as capture:
                pytask.console.print(pytask.Traceback(report.exc_info))
            s = capture.get()
            result = {
                "type": "task",
                "name": report.task.name.split("/")[-1],
                "outcome": str(report.outcome),
                "exc_info": s,
            }
        else:
            result = {
                "type": "task",
                "name": report.task.name.split("/")[-1],
                "outcome": str(report.outcome),
            }
        res = requests.post(
            "http://localhost:6000/pytask", json=result, timeout=0.00001
        )
    except requests.exceptions.ReadTimeout:
        pass
    except Exception:
        pass