zhou/pyflowx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

~clirunner

Browse Source
This commit is contained in:
gooker_young
2026-06-20 17:13:18 +08:00
parent e00868e3b1
commit 6d4b5e4a1f
10 changed files with 194 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pyflowx/cli/pymake.py
+2 -1
View File
@@ -403,7 +403,8 @@ def main():
pymake ca # 清理所有构建产物
"""
runner = px.CliRunner(
strategy=px.Strategy.SEQUENTIAL,
description="PyMake - Python 构建工具 (替代 Makefile)",
**_build_graphs(),
graphs=**_build_graphs(),
)
runner.run_cli()
src/pyflowx/context.py
+1 -3
View File
@@ -84,9 +84,7 @@ def build_call_args(
)
# 与本任务相关的上下文子集。
dep_context: Dict[str, Any] = {
name: context[name] for name in spec.depends_on if name in context
}
dep_context: Dict[str, Any] = {name: context[name] for name in spec.depends_on if name in context}
# 检测静态 kwargs 与依赖名的冲突。
collisions = set(spec.kwargs) & set(dep_context)
src/pyflowx/errors.py
+1 -3
View File
@@ -58,9 +58,7 @@ class TaskFailedError(PyFlowXError):
layer: Optional[int] = None,
) -> None:
location = f" (layer {layer})" if layer is not None else ""
super().__init__(
f"Task '{task}' failed after {attempts} attempt(s){location}: {cause}"
)
super().__init__(f"Task '{task}' failed after {attempts} attempt(s){location}: {cause}")
self.task = task
self.cause = cause
self.attempts = attempts
src/pyflowx/executors.py
+9 -27
View File
@@ -60,9 +60,7 @@ def _emit(
)
def _log_retry(
spec: TaskSpec[object], attempts: int, max_attempts: int, exc: BaseException
) -> None:
def _log_retry(spec: TaskSpec[object], attempts: int, max_attempts: int, exc: BaseException) -> None:
"""记录重试日志（sync 与 async 共享，便于测试覆盖）。"""
logger.warning(
"task %r failed (attempt %d/%d): %r; retrying",
@@ -154,9 +152,7 @@ async def _run_async_with_retry(
return spec.effective_fn(*args, **kwargs)
if spec.timeout is not None:
result.value = await asyncio.wait_for(
loop.run_in_executor(None, fn_call), timeout=spec.timeout
)
result.value = await asyncio.wait_for(loop.run_in_executor(None, fn_call), timeout=spec.timeout)
else:
result.value = await loop.run_in_executor(None, fn_call)
result.status = TaskStatus.SUCCESS
@@ -188,9 +184,7 @@ def _build_context(
global_context: Mapping[str, Any],
) -> Mapping[str, Any]:
"""将全局上下文限制为本任务的依赖。"""
return {
dep: global_context[dep] for dep in spec.depends_on if dep in global_context
}
return {dep: global_context[dep] for dep in spec.depends_on if dep in global_context}
def _execute_layer_sequential(
@@ -237,9 +231,7 @@ def _execute_layer_threaded(
if backend.has(name):
cached = backend.get(name)
context[name] = cached
result = TaskResult(
spec=graph.spec(name), status=TaskStatus.SKIPPED, value=cached
)
result = TaskResult(spec=graph.spec(name), status=TaskStatus.SKIPPED, value=cached)
report.results[name] = result
_emit(on_event, result)
else:
@@ -281,9 +273,7 @@ async def _execute_layer_async(
if backend.has(name):
cached = backend.get(name)
context[name] = cached
result = TaskResult(
spec=graph.spec(name), status=TaskStatus.SKIPPED, value=cached
)
result = TaskResult(spec=graph.spec(name), status=TaskStatus.SKIPPED, value=cached)
report.results[name] = result
_emit(on_event, result)
else:
@@ -346,9 +336,7 @@ def run(
不会被执行。
"""
if strategy not in ("sequential", "thread", "async"):
raise ValueError(
f"unknown strategy {strategy!r}; expected 'sequential', 'thread', or 'async'."
)
raise ValueError(f"unknown strategy {strategy!r}; expected 'sequential', 'thread', or 'async'.")
graph.validate()
layers = graph.layers()
@@ -365,9 +353,7 @@ def run(
if strategy == "sequential":
_drive_sequential(graph, layers, context, report, backend, on_event)
elif strategy == "thread":
_drive_threaded(
graph, layers, context, report, backend, on_event, max_workers
)
_drive_threaded(graph, layers, context, report, backend, on_event, max_workers)
else:
_drive_async(graph, layers, context, report, backend, on_event)
except TaskFailedError:
@@ -409,9 +395,7 @@ def _drive_threaded(
) -> None:
for idx, layer in enumerate(layers, 1):
workers = max_workers or max(1, min(32, len(layer)))
_execute_layer_threaded(
layer, graph, context, report, backend, idx, on_event, workers
)
_execute_layer_threaded(layer, graph, context, report, backend, idx, on_event, workers)
def _drive_async(
@@ -434,6 +418,4 @@ async def _async_drive(
on_event: Optional[EventCallback],
) -> None:
for idx, layer in enumerate(layers, 1):
await _execute_layer_async(
layer, graph, context, report, backend, idx, on_event
)
await _execute_layer_async(layer, graph, context, report, backend, idx, on_event)
src/pyflowx/graph.py
+6 -7
View File
@@ -10,12 +10,14 @@ from __future__ import annotations
import sys
from typing import Dict, Iterable, List, Mapping, Sequence, Set, Tuple
from typing_extensions import override
from .errors import CycleError, DuplicateTaskError, MissingDependencyError
from .task import TaskSpec
# graphlib 自 3.9 起进入标准库；3.8 回退到 backport。
if sys.version_info >= (3, 9): # pragma: no cover
import graphlib
import graphlib # pyright: ignore[reportUnreachable]
_TopologicalSorter = graphlib.TopologicalSorter
else: # pragma: no cover
@@ -157,9 +159,7 @@ class Graph:
for spec in self._specs.values():
if wanted & set(spec.tags):
pruned_deps = tuple(
d
for d in spec.depends_on
if d in self._specs and (wanted & set(self._specs[d].tags))
d for d in spec.depends_on if d in self._specs and (wanted & set(self._specs[d].tags))
)
kept.append(
TaskSpec(
@@ -217,9 +217,7 @@ class Graph:
valid = {"TD", "TB", "BT", "LR", "RL"}
orientation = orientation.upper()
if orientation not in valid:
raise ValueError(
f"Invalid orientation {orientation!r}; expected one of {sorted(valid)}."
)
raise ValueError(f"Invalid orientation {orientation!r}; expected one of {sorted(valid)}.")
lines: List[str] = [f"graph {orientation}"]
for name in self._specs:
lines.append(f' {name}["{name}"]')
@@ -238,6 +236,7 @@ class Graph:
out.append(f" Layer {layer_idx}: {layer}")
return "\n".join(out)
@override
def __repr__(self) -> str:
return f"Graph(tasks={len(self._specs)})"
src/pyflowx/runner.py
+9 -18
View File
@@ -16,9 +16,9 @@ import enum
import sys
from typing import Dict, List, Optional, Sequence
from ..errors import PyFlowXError
from ..executors import Strategy, run
from ..graph import Graph
from .errors import PyFlowXError
from .executors import Strategy, run
from .graph import Graph
__all__ = ["CliRunner", "CliExitCode"]
@@ -72,19 +72,10 @@ class CliRunner:
runner.run(["test", "--strategy", "sequential"])
"""
def __init__(
self,
*,
strategy: Strategy = "sequential",
description: str = "",
**graphs: Graph,
) -> None:
def __init__(self, *, strategy: Strategy = "sequential", description: str = "", graphs: Dict[str, Graph]) -> None:
if not graphs:
raise ValueError("CliRunner 至少需要一个命令 (通过关键字参数提供)")
# 校验所有值都是 Graph
for name, graph in graphs.items():
if not isinstance(graph, Graph):
raise TypeError(f"CliRunner 命令 {name!r} 的值必须是 Graph 实例, 实际是 {type(graph).__name__}")
self._graphs: Dict[str, Graph] = dict(graphs)
self._strategy: Strategy = strategy
self._description: str = description
@@ -139,23 +130,23 @@ class CliRunner:
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=self._format_commands_help(),
)
parser.add_argument(
_ = parser.add_argument(
"command",
nargs="?",
help="要执行的命令",
)
parser.add_argument(
_ = parser.add_argument(
"--strategy",
choices=["sequential", "thread", "async"],
default=self._strategy,
help="执行策略 (默认: %(default)s)",
)
parser.add_argument(
_ = parser.add_argument(
"--dry-run",
action="store_true",
help="只打印执行计划, 不实际运行",
)
parser.add_argument(
_ = parser.add_argument(
"--list",
action="store_true",
help="列出所有可用命令",
src/pyflowx/task.py
+1 -3
View File
@@ -227,9 +227,7 @@ class TaskSpec(Generic[T]):
if callable(cmd):
return cmd # type: ignore[return-value]
raise TypeError(
f"TaskSpec '{self.name}': 不支持的 cmd 类型 {type(cmd).__name__}"
)
raise TypeError(f"TaskSpec '{self.name}': 不支持的 cmd 类型 {type(cmd).__name__}")
def should_execute(self) -> bool:
"""检查任务是否应该执行.
tests/test_cli_runner.py → tests/test_runner.py
+47 -81
View File
@@ -9,8 +9,8 @@ from unittest.mock import patch
import pytest
import pyflowx as px
from pyflowx.cli import CliExitCode, CliRunner
from pyflowx.errors import PyFlowXError, TaskFailedError
from pyflowx import CliExitCode, CliRunner
from pyflowx.errors import TaskFailedError
# 跨平台的 echo 命令
if sys.platform == "win32":
@@ -29,24 +29,20 @@ def _echo_graph(name: str = "echo_task", msg: str = "hello") -> px.Graph:
def _failing_graph() -> px.Graph:
"""构造一个必定失败的单任务图."""
return px.Graph.from_specs(
[
px.TaskSpec(
"fail",
cmd=["python", "-c", "import sys; sys.exit(1)"],
)
]
)
return px.Graph.from_specs([
px.TaskSpec(
"fail",
cmd=["python", "-c", "import sys; sys.exit(1)"],
)
])
def _multi_task_graph() -> px.Graph:
"""构造一个带依赖的多任务图."""
return px.Graph.from_specs(
[
px.TaskSpec("a", cmd=[*ECHO_CMD, "a"]),
px.TaskSpec("b", cmd=[*ECHO_CMD, "b"], depends_on=("a",)),
]
)
return px.Graph.from_specs([
px.TaskSpec("a", cmd=[*ECHO_CMD, "a"]),
px.TaskSpec("b", cmd=[*ECHO_CMD, "b"], depends_on=("a",)),
])
# ---------------------------------------------------------------------- #
@@ -218,17 +214,13 @@ class TestCliRunnerParser:
class TestCliRunnerRunSuccess:
"""测试 CliRunner.run 的成功执行路径."""
def test_run_valid_command_returns_zero(
self, capsys: pytest.CaptureFixture[str]
) -> None:
def test_run_valid_command_returns_zero(self, capsys: pytest.CaptureFixture[str]) -> None:
"""有效命令执行成功应返回 0."""
runner = px.CliRunner(echo=_echo_graph())
exit_code = runner.run(["echo"])
assert exit_code == CliExitCode.SUCCESS.value
def test_run_executes_correct_graph(
self, capsys: pytest.CaptureFixture[str]
) -> None:
def test_run_executes_correct_graph(self, capsys: pytest.CaptureFixture[str]) -> None:
"""应执行用户指定的命令对应的图."""
executed: List[str] = []
@@ -272,9 +264,7 @@ class TestCliRunnerRunSuccess:
class TestCliRunnerRunFailure:
"""测试 CliRunner.run 的失败执行路径."""
def test_run_unknown_command_returns_failure(
self, capsys: pytest.CaptureFixture[str]
) -> None:
def test_run_unknown_command_returns_failure(self, capsys: pytest.CaptureFixture[str]) -> None:
"""未知命令应返回 1 并打印错误."""
runner = px.CliRunner(clean=_echo_graph())
exit_code = runner.run(["unknown"])
@@ -283,9 +273,7 @@ class TestCliRunnerRunFailure:
assert "未知命令" in captured.err
assert "clean" in captured.err
def test_run_no_command_returns_failure(
self, capsys: pytest.CaptureFixture[str]
) -> None:
def test_run_no_command_returns_failure(self, capsys: pytest.CaptureFixture[str]) -> None:
"""无命令时应返回 1 并打印帮助."""
runner = px.CliRunner(clean=_echo_graph())
exit_code = runner.run([])
@@ -293,17 +281,13 @@ class TestCliRunnerRunFailure:
captured = capsys.readouterr()
assert "可用命令" in captured.out or "可用命令" in captured.err
def test_run_failing_task_returns_failure(
self, capsys: pytest.CaptureFixture[str]
) -> None:
def test_run_failing_task_returns_failure(self, capsys: pytest.CaptureFixture[str]) -> None:
"""任务失败时应返回 1."""
runner = px.CliRunner(fail=_failing_graph())
exit_code = runner.run(["fail"])
assert exit_code == CliExitCode.FAILURE.value
def test_run_failing_task_prints_error(
self, capsys: pytest.CaptureFixture[str]
) -> None:
def test_run_failing_task_prints_error(self, capsys: pytest.CaptureFixture[str]) -> None:
"""任务失败时应打印错误信息."""
runner = px.CliRunner(fail=_failing_graph())
runner.run(["fail"])
@@ -337,9 +321,7 @@ class TestCliRunnerList:
assert "build" in captured.out
assert "test" in captured.out
def test_list_does_not_execute_any_graph(
self, capsys: pytest.CaptureFixture[str]
) -> None:
def test_list_does_not_execute_any_graph(self, capsys: pytest.CaptureFixture[str]) -> None:
"""--list 不应执行任何图."""
executed: List[str] = []
@@ -357,31 +339,27 @@ class TestCliRunnerList:
class TestCliRunnerErrorHandling:
"""测试错误处理."""
def test_keyboard_interrupt_returns_130(
self, capsys: pytest.CaptureFixture[str]
) -> None:
def test_keyboard_interrupt_returns_130(self, capsys: pytest.CaptureFixture[str]) -> None:
"""KeyboardInterrupt 应返回 130."""
runner = px.CliRunner(echo=_echo_graph())
def raise_interrupt(*args: Any, **kwargs: Any) -> None:
raise KeyboardInterrupt
with patch("pyflowx.cli.runner.run", side_effect=raise_interrupt):
with patch("pyflowx.runner.run", side_effect=raise_interrupt):
exit_code = runner.run(["echo"])
assert exit_code == CliExitCode.INTERRUPTED.value
captured = capsys.readouterr()
assert "取消" in captured.err
def test_pyflowx_error_returns_failure(
self, capsys: pytest.CaptureFixture[str]
) -> None:
def test_pyflowx_error_returns_failure(self, capsys: pytest.CaptureFixture[str]) -> None:
"""PyFlowXError 应返回 1."""
runner = px.CliRunner(echo=_echo_graph())
def raise_error(*args: Any, **kwargs: Any) -> None:
raise TaskFailedError("echo", RuntimeError("boom"), 1)
with patch("pyflowx.cli.runner.run", side_effect=raise_error):
with patch("pyflowx.runner.run", side_effect=raise_error):
exit_code = runner.run(["echo"])
assert exit_code == CliExitCode.FAILURE.value
captured = capsys.readouterr()
@@ -398,7 +376,7 @@ class TestCliRunnerErrorHandling:
def raise_custom(*args: Any, **kwargs: Any) -> None:
raise CustomError("unexpected")
with patch("pyflowx.cli.runner.run", side_effect=raise_custom):
with patch("pyflowx.runner.run", side_effect=raise_custom):
with pytest.raises(CustomError):
runner.run(["echo"])
@@ -423,9 +401,7 @@ class TestCliRunnerRunCli:
runner.run_cli(["fail"])
assert exc_info.value.code == CliExitCode.FAILURE.value
def test_run_cli_no_args_uses_sys_argv(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
def test_run_cli_no_args_uses_sys_argv(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""run_cli 无参数时应使用 sys.argv."""
monkeypatch.setattr(sys, "argv", ["pymake", "echo"])
runner = px.CliRunner(echo=_echo_graph())
@@ -462,30 +438,26 @@ class TestCliRunnerIntegration:
def test_condition_skipped_command_succeeds(self) -> None:
"""条件不满足时任务跳过, 整体仍成功."""
graph = px.Graph.from_specs(
[
px.TaskSpec(
"skip_me",
cmd=[*ECHO_CMD, "should not run"],
conditions=(lambda: False,),
),
]
)
graph = px.Graph.from_specs([
px.TaskSpec(
"skip_me",
cmd=[*ECHO_CMD, "should not run"],
conditions=(lambda: False,),
),
])
runner = px.CliRunner(skip=graph)
exit_code = runner.run(["skip"])
assert exit_code == CliExitCode.SUCCESS.value
def test_condition_met_command_succeeds(self) -> None:
"""条件满足时任务执行, 整体成功."""
graph = px.Graph.from_specs(
[
px.TaskSpec(
"run_me",
cmd=[*ECHO_CMD, "should run"],
conditions=(lambda: True,),
),
]
)
graph = px.Graph.from_specs([
px.TaskSpec(
"run_me",
cmd=[*ECHO_CMD, "should run"],
conditions=(lambda: True,),
),
])
runner = px.CliRunner(run=graph)
exit_code = runner.run(["run"])
assert exit_code == CliExitCode.SUCCESS.value
@@ -501,14 +473,12 @@ class TestCliRunnerIntegration:
return fn
graph = px.Graph.from_specs(
[
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b"), depends_on=("a",)),
px.TaskSpec("c", make("c"), depends_on=("a",)),
px.TaskSpec("d", make("d"), depends_on=("b", "c")),
]
)
graph = px.Graph.from_specs([
px.TaskSpec("a", make("a")),
px.TaskSpec("b", make("b"), depends_on=("a",)),
px.TaskSpec("c", make("c"), depends_on=("a",)),
px.TaskSpec("d", make("d"), depends_on=("b", "c")),
])
runner = px.CliRunner(diamond=graph)
exit_code = runner.run(["diamond"])
assert exit_code == CliExitCode.SUCCESS.value
@@ -518,9 +488,7 @@ class TestCliRunnerIntegration:
"""混合 fn 和 cmd 的命令应都能执行."""
runner = px.CliRunner(
fn_cmd=px.Graph.from_specs([px.TaskSpec("fn", fn=lambda: "fn-result")]),
cmd_cmd=px.Graph.from_specs(
[px.TaskSpec("cmd", cmd=[*ECHO_CMD, "cmd-result"])]
),
cmd_cmd=px.Graph.from_specs([px.TaskSpec("cmd", cmd=[*ECHO_CMD, "cmd-result"])]),
)
assert runner.run(["fn_cmd"]) == CliExitCode.SUCCESS.value
assert runner.run(["cmd_cmd"]) == CliExitCode.SUCCESS.value
@@ -536,9 +504,7 @@ class TestCliRunnerIntegration:
else:
ls_cmd = ["ls"]
graph = px.Graph.from_specs(
[px.TaskSpec("ls", cmd=ls_cmd, cwd=Path(tmpdir))]
)
graph = px.Graph.from_specs([px.TaskSpec("ls", cmd=ls_cmd, cwd=Path(tmpdir))])
runner = px.CliRunner(ls=graph)
exit_code = runner.run(["ls"])
assert exit_code == CliExitCode.SUCCESS.value
typings/graphlib/__init__.pyi
+6
View File
@@ -0,0 +1,6 @@
"""
This type stub file was generated by pyright.
"""
from .graphlib import *
typings/graphlib/graphlib.pyi
+112
View File
@@ -0,0 +1,112 @@
"""
This type stub file was generated by pyright.
"""
__all__ = ["TopologicalSorter", "CycleError"]
_NODE_OUT = ...
_NODE_DONE = ...
class _NodeInfo:
__slots__ = ...
def __init__(self, node) -> None:
...
class CycleError(ValueError):
"""Subclass of ValueError raised by TopologicalSorterif cycles exist in the graph
If multiple cycles exist, only one undefined choice among them will be reported
and included in the exception. The detected cycle can be accessed via the second
element in the *args* attribute of the exception instance and consists in a list
of nodes, such that each node is, in the graph, an immediate predecessor of the
next node in the list. In the reported list, the first and the last node will be
the same, to make it clear that it is cyclic.
"""
...
class TopologicalSorter:
"""Provides functionality to topologically sort a graph of hashable nodes"""
def __init__(self, graph=...) -> None:
...
def add(self, node, *predecessors) -> None:
"""Add a new node and its predecessors to the graph.
Both the *node* and all elements in *predecessors* must be hashable.
If called multiple times with the same node argument, the set of dependencies
will be the union of all dependencies passed in.
It is possible to add a node with no dependencies (*predecessors* is not provided)
as well as provide a dependency twice. If a node that has not been provided before
is included among *predecessors* it will be automatically added to the graph with
no predecessors of its own.
Raises ValueError if called after "prepare".
"""
...
def prepare(self) -> None:
"""Mark the graph as finished and check for cycles in the graph.
If any cycle is detected, "CycleError" will be raised, but "get_ready" can
still be used to obtain as many nodes as possible until cycles block more
progress. After a call to this function, the graph cannot be modified and
therefore no more nodes can be added using "add".
"""
...
def get_ready(self) -> tuple[Any, ...]:
"""Return a tuple of all the nodes that are ready.
Initially it returns all nodes with no predecessors; once those are marked
as processed by calling "done", further calls will return all new nodes that
have all their predecessors already processed. Once no more progress can be made,
empty tuples are returned.
Raises ValueError if called without calling "prepare" previously.
"""
...
def is_active(self) -> bool:
"""Return True if more progress can be made and ``False`` otherwise.
Progress can be made if cycles do not block the resolution and either there
are still nodes ready that haven't yet been returned by "get_ready" or the
number of nodes marked "done" is less than the number that have been returned
by "get_ready".
Raises ValueError if called without calling "prepare" previously.
"""
...
def __bool__(self) -> bool:
...
def done(self, *nodes) -> None:
"""Marks a set of nodes returned by "get_ready" as processed.
This method unblocks any successor of each node in *nodes* for being returned
in the future by a a call to "get_ready"
Raises :exec:`ValueError` if any node in *nodes* has already been marked as
processed by a previous call to this method, if a node was not added to the
graph by using "add" or if called without calling "prepare" previously or if
node has not yet been returned by "get_ready".
"""
...
def static_order(self) -> Generator[Any, Any, None]:
"""Returns an iterable of nodes in a topological order.
The particular order that is returned may depend on the specific
order in which the items were inserted in the graph.
Using this method does not require to call "prepare" or "done". If any
cycle is detected, :exc:`CycleError` will be raised.
"""
...