From ae4cc3aeebe876f0d06b50da859ca514c4b0c09e Mon Sep 17 00:00:00 2001 From: gooker_young Date: Wed, 8 Jul 2026 14:56:54 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=20px.sh=20=E5=91=BD?= =?UTF-8?q?=E4=BB=A4=E6=89=A7=E8=A1=8C=E5=B0=81=E8=A3=85=20=E2=80=94=20?= =?UTF-8?q?=E6=94=AF=E6=8C=81=20list[str]=EF=BC=88=E7=9B=B4=E6=8E=A5?= =?UTF-8?q?=E6=89=A7=E8=A1=8C=EF=BC=89/=20str=EF=BC=88shell=3DTrue?= =?UTF-8?q?=EF=BC=89=EF=BC=8C=E7=BB=9F=E4=B8=80=E4=B8=AD=E6=96=87=E9=94=99?= =?UTF-8?q?=E8=AF=AF=E5=A4=84=E7=90=86=EF=BC=8C=E5=A4=B1=E8=B4=A5=E8=BF=94?= =?UTF-8?q?=E5=9B=9E=20None=20=E4=B8=8D=E6=8A=9B=E5=BC=82=E5=B8=B8?= =?UTF-8?q?=EF=BC=9Bsglang.py=20=E9=A6=96=E4=B8=AA=E8=BF=81=E7=A7=BB?= =?UTF-8?q?=E7=A4=BA=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .trae/docs/iter-28-px-sh.md | 70 ++++++++++++++ src/pyflowx/__init__.py | 3 + src/pyflowx/ops/infra/sglang.py | 15 +-- src/pyflowx/shell.py | 75 +++++++++++++++ tests/cli/test_llm.py | 4 +- tests/test_shell.py | 164 ++++++++++++++++++++++++++++++++ 6 files changed, 316 insertions(+), 15 deletions(-) create mode 100644 .trae/docs/iter-28-px-sh.md create mode 100644 src/pyflowx/shell.py create mode 100644 tests/test_shell.py diff --git a/.trae/docs/iter-28-px-sh.md b/.trae/docs/iter-28-px-sh.md new file mode 100644 index 0000000..00c37b8 --- /dev/null +++ b/.trae/docs/iter-28-px-sh.md @@ -0,0 +1,70 @@ +# iter-28：px.sh 命令执行封装 + +## 本轮目标 + +ops/ 目录下 57 处 `subprocess.run` 调用存在重复的 `try/except CalledProcessError + +FileNotFoundError + print` 模式。封装一个轻量的 `px.sh` 辅助函数，统一错误处理 +与中文输出，支持 `list[str]`（直接执行）和 `str`（`shell=True`）两种输入。 + +## 改动文件清单 + +| 文件 | 变更 | +|------|------| +| `src/pyflowx/shell.py` | 新建模块，实现 `sh()` 函数 | +| `src/pyflowx/__init__.py` | 导出 `sh`，更新 docstring | +| `tests/test_shell.py` | 新建，16 个测试（list/str/mocked/exported） | +| `src/pyflowx/ops/infra/sglang.py` | 重构为使用 `px.sh`，移除 try/except 样板 | +| `tests/cli/test_llm.py` | 适配 sglang.py 重构后的错误信息格式 | + +## 关键决策与依据 + +### 命名与定位 + +- **`px.sh` 而非 `px.run`**：`px.run` 已被占用（执行 Graph DAG），`sh` 简短易记， + 语义清晰（"执行 shell 命令"）。 +- **轻量封装，不替代所有 `subprocess.run`**：需要 `capture_output` + 手动检查 + 返回码等精细控制的场景（如 `lscalc.py`/`piptool.py`/`which.py`）保留原样。 + `px.sh` 定位为"简单执行 + 报错"场景的快捷方式。 + +### 输入类型 + +- **`list[str]` → `shell=False`**：直接执行，安全无注入风险。 +- **`str` → `shell=True`**：支持管道/重定向等 shell 语法；ops 工具场景下命令可信。 +- **不支持 `func`**：ops 工具本身就是 Python 函数，直接 `func()` 调用即可， + 封装反而增加复杂度。 + +### 返回值设计 + +- **成功返回 `CompletedProcess[str]`**：`text=True` 始终启用，输出为字符串。 +- **失败返回 `None` + 打印中文错误**：不抛异常，调用方用 `result is None` 判断。 + 统一 `CalledProcessError`（打印 returncode）和 `FileNotFoundError`（打印"命令未找到"）。 +- **`capture` 参数**：`False`（默认）输出到终端；`True` 捕获 stdout/stderr。 + +### sglang.py 迁移示例 + +重构前（12 行 try/except 样板）： +```python +try: + subprocess.run(["uv", "install", "sglang[all]"], check=True) +except subprocess.CalledProcessError as e: + print(f"安装失败 (returncode={e.returncode}): {e}") +except FileNotFoundError: + print("uv 命令未找到，请确认 uv 已安装并在 PATH 中") +``` + +重构后（1 行）： +```python +px.sh(["uv", "install", "sglang[all]"], label="安装 sglang") +``` + +## 验证结果 + +- pytest：1721 passed（新增 16 个 shell 测试） +- coverage：97.20%（shell.py 100%、sglang.py 100%） +- ruff：All checks passed +- pyrefly：0 errors + +## 遗留事项 + +- 其余 ~55 处 `subprocess.run` 可在后续迭代中按需迁移到 `px.sh`。 + 优先迁移"简单执行 + 报错"模式的调用；`capture_output` 等精细控制场景保留原样。 diff --git a/src/pyflowx/__init__.py b/src/pyflowx/__init__.py index ae190ea..36ff243 100644 --- a/src/pyflowx/__init__.py +++ b/src/pyflowx/__init__.py @@ -16,6 +16,7 @@ * :func:`compose` —— 编程式组合多图。 * :func:`switch` / :func:`branch` —— 条件分支 DAG 构造器。 * :func:`task_template` —— 批量生成相似 TaskSpec 的工厂。 +* :func:`sh` —— Shell 命令执行辅助（支持 ``list[str]`` / ``str``，统一中文错误处理）。 * 状态后端：:class:`StateBackend`、:class:`MemoryBackend`、:class:`JSONBackend`、:class:`SQLiteBackend`。 快速上手 @@ -92,6 +93,7 @@ from .profiling import ProfileReport, TaskProfile from .progress import ProgressCallback, RichProgressMonitor from .report import RunReport from .runner import CliExitCode, CliRunner +from .shell import sh from .storage import JSONBackend, MemoryBackend, SQLiteBackend, StateBackend from .task import ( CacheKeyFn, @@ -215,6 +217,7 @@ __all__ = [ "run_command", "run_iter", "run_tool", + "sh", "start_metrics_server", "switch", "task", diff --git a/src/pyflowx/ops/infra/sglang.py b/src/pyflowx/ops/infra/sglang.py index afe796a..5fbefdd 100644 --- a/src/pyflowx/ops/infra/sglang.py +++ b/src/pyflowx/ops/infra/sglang.py @@ -6,7 +6,6 @@ from __future__ import annotations import shutil -import subprocess from pathlib import Path import pyflowx as px @@ -29,12 +28,7 @@ def install_sglang() -> None: return print("正在安装 sglang[all]...") - try: - subprocess.run(["uv", "install", "sglang[all]"], check=True) - except subprocess.CalledProcessError as e: - print(f"安装失败 (returncode={e.returncode}): {e}") - except FileNotFoundError: - print("uv 命令未找到，请确认 uv 已安装并在 PATH 中") + px.sh(["uv", "install", "sglang[all]"], label="安装 sglang") @px.tool("sglang", subcommand="run", help="启动 SGLang 服务", needs=["install"]) @@ -89,9 +83,4 @@ def run_sglang( log_level, ] print(f"启动 SGLang: {model_dir} (port={port}, ctx={ctx_len}, mem={mem_fraction})") - try: - subprocess.run(cmd, check=True) - except subprocess.CalledProcessError as e: - print(f"SGLang 启动失败 (returncode={e.returncode}): {e}") - except FileNotFoundError: - print(f"{python_bin} 命令未找到，请确认 Python 已安装并在 PATH 中") + px.sh(cmd, label="SGLang 启动") diff --git a/src/pyflowx/shell.py b/src/pyflowx/shell.py new file mode 100644 index 0000000..17bf2ba --- /dev/null +++ b/src/pyflowx/shell.py @@ -0,0 +1,75 @@ +"""Shell 命令执行辅助。 + +提供 :func:`sh` —— 对 :func:`subprocess.run` 的轻量封装，统一错误处理 +与中文输出，支持 ``list[str]``（直接执行）和 ``str``（``shell=True``）两种输入。 + +典型用法:: + + import pyflowx as px + + px.sh(["git", "add", "."]) # list[str]，直接执行 + px.sh("git add . | tee log.txt") # str，shell=True（支持管道/重定向） + result = px.sh("echo hello", capture=True) + if result is not None: + print(result.stdout.strip()) + +设计定位：简单"执行 + 报错"场景的快捷方式。需要 ``capture_output`` + +手动检查返回码等精细控制时，直接使用 :func:`subprocess.run`。 +""" + +from __future__ import annotations + +import subprocess + +__all__ = ["sh"] + + +def sh( + cmd: list[str] | str, + *, + capture: bool = False, + label: str = "命令", +) -> subprocess.CompletedProcess[str] | None: + """执行 shell 命令，统一错误处理与中文输出。 + + ``list[str]`` 直接执行；``str`` 走 ``shell=True``（支持管道/重定向等语法）。 + 成功返回 :class:`subprocess.CompletedProcess`；失败时打印中文错误信息 + 并返回 ``None``（不抛异常），调用方用 ``result is None`` 判断是否失败。 + + Parameters + ---------- + cmd : list[str] | str + 命令列表（直接执行）或 shell 字符串（``shell=True`` 执行） + capture : bool + 为 ``True`` 时捕获 stdout/stderr（不直接输出到终端）， + 可从返回值的 ``.stdout`` / ``.stderr`` 读取（默认 ``False``） + label : str + 命令标签，用于错误信息中标识命令用途（默认 ``"命令"``） + + Returns + ------- + subprocess.CompletedProcess[str] | None + 成功时返回执行结果；失败时返回 ``None`` + + Examples + -------- + >>> px.sh(["ls", "-la"]) # 列出当前目录 + >>> px.sh("cat file.txt | grep pattern") # shell 管道 + >>> result = px.sh("pip --version", capture=True, label="pip") + >>> if result is not None: + ... print(result.stdout) + """ + try: + return subprocess.run( + cmd, + shell=isinstance(cmd, str), + capture_output=capture, + text=True, + check=True, + ) + except subprocess.CalledProcessError as e: + print(f"{label}失败 (returncode={e.returncode}): {e}") + return None + except FileNotFoundError: + print(f"{label}失败: 命令未找到，请确认已安装并在 PATH 中") + return None diff --git a/tests/cli/test_llm.py b/tests/cli/test_llm.py index 3f341a7..b9e9cf9 100644 --- a/tests/cli/test_llm.py +++ b/tests/cli/test_llm.py @@ -130,7 +130,7 @@ class TestInstallSglang: monkeypatch.setattr(subprocess, "run", lambda *_, **__: (_ for _ in ()).throw(err)) install_sglang() captured = capsys.readouterr() - assert "安装失败" in captured.out + assert "安装 sglang失败" in captured.out assert "returncode=1" in captured.out def test_install_file_not_found_prints_no_uv( @@ -141,7 +141,7 @@ class TestInstallSglang: monkeypatch.setattr(subprocess, "run", lambda *_, **__: (_ for _ in ()).throw(FileNotFoundError())) install_sglang() captured = capsys.readouterr() - assert "uv 命令未找到" in captured.out + assert "命令未找到" in captured.out # ---------------------------------------------------------------------- # diff --git a/tests/test_shell.py b/tests/test_shell.py new file mode 100644 index 0000000..8867b29 --- /dev/null +++ b/tests/test_shell.py @@ -0,0 +1,164 @@ +"""px.sh 命令执行封装的测试。""" + +from __future__ import annotations + +import subprocess +from typing import Any + +import pyflowx as px +from pyflowx.shell import sh + + +class TestShListInput: + """list[str] 输入（直接执行，非 shell）。""" + + def test_sh_list_success(self) -> None: + """list[str] 命令成功执行，返回 CompletedProcess。""" + result = sh(["true"]) + assert result is not None + assert result.returncode == 0 + + def test_sh_list_success_with_capture(self) -> None: + """capture=True 时捕获 stdout。""" + result = sh(["echo", "hello"], capture=True, label="echo") + assert result is not None + assert result.stdout == "hello\n" + + def test_sh_list_no_capture_stdout_is_none(self) -> None: + """capture=False 时 stdout 为 None（输出到终端）。""" + result = sh(["true"]) + assert result is not None + assert result.stdout is None + + def test_sh_list_failure_returns_none(self, capsys: Any) -> None: + """list[str] 命令失败（非零返回码）返回 None + 打印中文错误。""" + result = sh(["false"], label="测试命令") + assert result is None + captured = capsys.readouterr() + assert "测试命令失败" in captured.out + assert "returncode=" in captured.out + + def test_sh_list_file_not_found_returns_none(self, capsys: Any) -> None: + """list[str] 命令不存在返回 None + 打印中文错误。""" + result = sh(["this_command_does_not_exist_xyz"], label="缺失命令") + assert result is None + captured = capsys.readouterr() + assert "缺失命令失败" in captured.out + assert "命令未找到" in captured.out + + +class TestShStrInput: + """str 输入（shell=True）。""" + + def test_sh_str_success(self) -> None: + """str 命令成功执行（shell=True）。""" + result = sh("true") + assert result is not None + assert result.returncode == 0 + + def test_sh_str_with_pipe(self) -> None: + """str 支持 shell 管道语法。""" + result = sh("echo hello | tr a-z A-Z", capture=True, label="管道") + assert result is not None + assert result.stdout == "HELLO\n" + + def test_sh_str_with_redirect(self, tmp_path: Any) -> None: + """str 支持 shell 重定向语法。""" + out_file = tmp_path / "out.txt" + sh(f"echo content > {out_file}") + assert out_file.read_text() == "content\n" + + def test_sh_str_failure_returns_none(self, capsys: Any) -> None: + """str 命令失败返回 None + 打印中文错误。""" + result = sh("false", label="shell 命令") + assert result is None + captured = capsys.readouterr() + assert "shell 命令失败" in captured.out + + def test_sh_str_capture_stderr(self) -> None: + """capture=True 时捕获 stderr。""" + result = sh("echo err >&2", capture=True, label="stderr") + assert result is not None + assert result.stderr == "err\n" + + +class TestShMocked: + """用 monkeypatch mock subprocess.run 验证参数传递。""" + + def test_sh_list_passes_shell_false(self, monkeypatch: Any) -> None: + """list[str] 输入时 shell=False。""" + captured: dict[str, Any] = {} + + def fake_run(cmd: Any, **kwargs: Any) -> subprocess.CompletedProcess[str]: + captured["cmd"] = cmd + captured["shell"] = kwargs.get("shell") + captured["capture_output"] = kwargs.get("capture_output") + captured["text"] = kwargs.get("text") + captured["check"] = kwargs.get("check") + return subprocess.CompletedProcess(cmd, 0) + + monkeypatch.setattr(subprocess, "run", fake_run) + sh(["ls", "-la"]) + assert captured["cmd"] == ["ls", "-la"] + assert captured["shell"] is False + assert captured["capture_output"] is False + assert captured["text"] is True + assert captured["check"] is True + + def test_sh_str_passes_shell_true(self, monkeypatch: Any) -> None: + """str 输入时 shell=True。""" + captured: dict[str, Any] = {} + + def fake_run(cmd: Any, **kwargs: Any) -> subprocess.CompletedProcess[str]: + captured["cmd"] = cmd + captured["shell"] = kwargs.get("shell") + return subprocess.CompletedProcess(cmd, 0) + + monkeypatch.setattr(subprocess, "run", fake_run) + sh("ls -la | grep foo") + assert captured["cmd"] == "ls -la | grep foo" + assert captured["shell"] is True + + def test_sh_capture_passes_capture_output_true(self, monkeypatch: Any) -> None: + """capture=True 时 capture_output=True。""" + captured: dict[str, Any] = {} + + def fake_run(cmd: Any, **kwargs: Any) -> subprocess.CompletedProcess[str]: + captured["capture_output"] = kwargs.get("capture_output") + return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + sh(["echo"], capture=True) + assert captured["capture_output"] is True + + def test_sh_called_process_error_returns_none(self, monkeypatch: Any, capsys: Any) -> None: + """CalledProcessError 被捕获，打印中文错误并返回 None。""" + def fake_run(cmd: Any, **kwargs: Any) -> subprocess.CompletedProcess[str]: + raise subprocess.CalledProcessError(returncode=1, cmd=cmd) + + monkeypatch.setattr(subprocess, "run", fake_run) + result = sh(["failing"], label="自定义标签") + assert result is None + captured = capsys.readouterr() + assert "自定义标签失败" in captured.out + assert "returncode=1" in captured.out + + def test_sh_file_not_found_error_returns_none(self, monkeypatch: Any, capsys: Any) -> None: + """FileNotFoundError 被捕获，打印中文错误并返回 None。""" + def fake_run(cmd: Any, **kwargs: Any) -> subprocess.CompletedProcess[str]: + raise FileNotFoundError("[Errno 2] No such file") + + monkeypatch.setattr(subprocess, "run", fake_run) + result = sh(["missing"], label="缺失工具") + assert result is None + captured = capsys.readouterr() + assert "缺失工具失败" in captured.out + assert "命令未找到" in captured.out + + +class TestShExported: + """验证 px.sh 顶层导出。""" + + def test_sh_is_exported_from_px(self) -> None: + """px.sh 等价于 pyflowx.shell.sh。""" + assert px.sh is sh