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

chore: 升级项目适配 Python 3.10+ 并重构代码

Browse Source 
1.  移除对 Python 3.8/3.9 的支持，更新 tox、pyproject.toml 配置
2.  替换 typing 导入为 collections.abc 标准库类型
3.  重构 CLI 系统从 argparse 迁移到 typer
4.  优化代码格式与类型注解，修复多处类型兼容问题
5.  更新依赖声明，移除 graphlib_backport 兼容包
This commit is contained in:
gooker_young
2026-07-06 22:14:59 +08:00
parent d493c6233e
commit fca6f17a5c
30 changed files with 825 additions and 4796 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+5 -6
View File
@@ -7,12 +7,11 @@ classifiers = [
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: 3.14",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Topic :: Software Development :: Libraries :: Application Frameworks",
]
dependencies = [
"graphlib_backport >= 1.0.0; python_version < '3.9'",
"rich>=13.7.0",
"typer>=0.24.0",
"typing-extensions>=4.13.2; python_version < '3.13'",
]
description = "Lightweight, type-safe DAG task scheduler with multi-strategy execution."
@@ -20,7 +19,7 @@ keywords = ["async", "dag", "scheduler", "task", "workflow"]
license = { text = "MIT" }
name = "pyflowx"
readme = "README.md"
requires-python = ">=3.8"
requires-python = ">=3.10"
version = "0.4.8"
[project.scripts]
@@ -98,7 +97,7 @@ markers = ["slow: marks tests as slow (deselect with
# Ruff 配置 - 与 .pre-commit-config.yaml 保持一致
[tool.ruff]
line-length = 120
target-version = "py38"
target-version = "py310"
[tool.ruff.lint]
ignore = [
@@ -137,4 +136,4 @@ select = [
[tool.pyrefly]
preset = "strict"
project-includes = ["**/*.ipynb", "**/*.py*"]
python-version = "3.8"
python-version = "3.10"
src/pyflowx/cli/emlmanager.py
+3 -3
View File
@@ -154,7 +154,7 @@ class EmailDatabase:
cursor.execute(query, (f"%{keyword}%", f"%{keyword}%", f"%{keyword}%", f"%{keyword}%", limit, offset))
columns = [description[0] for description in cursor.description]
return [dict(zip(columns, row)) for row in cursor.fetchall()]
return [dict(zip(columns, row, strict=False)) for row in cursor.fetchall()]
def get_grouped_emails(self) -> dict[str, list[dict[str, Any]]]:
"""获取按主题分组的邮件."""
@@ -165,7 +165,7 @@ class EmailDatabase:
cursor.execute(f"SELECT * FROM {TABLE_NAME} ORDER BY subject, date_parsed DESC")
columns = [description[0] for description in cursor.description]
emails = [dict(zip(columns, row)) for row in cursor.fetchall()]
emails = [dict(zip(columns, row, strict=False)) for row in cursor.fetchall()]
# 按主题分组
grouped: dict[str, list[dict[str, Any]]] = {}
@@ -602,7 +602,7 @@ class EmlManagerHandler(BaseHTTPRequestHandler):
self._send_json_response({"error": "邮件不存在"}, 404)
return
email_data = dict(zip(columns, row))
email_data = dict(zip(columns, row, strict=False))
self._send_json_response({"email": email_data})
def _api_get_grouped_emails(self) -> None:
src/pyflowx/cli/pf.py
+1 -1
View File
@@ -16,7 +16,7 @@ from __future__ import annotations
import contextlib
import importlib
import sys
from typing import Sequence
from collections.abc import Sequence
from pyflowx.tools import _TOOL_REGISTRY, run_tool
src/pyflowx/command.py
+2 -2
View File
@@ -10,7 +10,7 @@ from __future__ import annotations
import os
import subprocess
from typing import Any, List, Union, cast
from typing import Any, cast
from .task import TaskSpec
@@ -70,7 +70,7 @@ def run_command(spec: TaskSpec[Any]) -> Any: # noqa: PLR0912
try:
result = subprocess.run(
cast(Union[str, List[str]], cmd),
cast(str | list[str], cmd),
shell=not is_list,
cwd=cwd,
env=run_env,
src/pyflowx/conditions.py
+2 -1
View File
@@ -16,8 +16,9 @@ import os
import shutil
import subprocess
import sys
from collections.abc import Callable
from pathlib import Path
from typing import Any, Callable
from typing import Any
from .task import Condition, Context
src/pyflowx/context.py
+2 -1
View File
@@ -16,8 +16,9 @@ DAG 库中泛滥的样板包装器。
from __future__ import annotations
import inspect
from collections.abc import Mapping
from functools import lru_cache
from typing import Any, Mapping
from typing import Any
from .errors import InjectionError
from .task import Context, TaskSpec
src/pyflowx/errors.py
+2 -1
View File
@@ -6,7 +6,8 @@
from __future__ import annotations
from typing import Any, Iterable
from collections.abc import Iterable
from typing import Any
class PyFlowXError(Exception):
src/pyflowx/executors.py
+3 -2
View File
@@ -48,8 +48,9 @@ import inspect
import logging
import threading
import time
from collections.abc import Awaitable, Callable, Mapping
from datetime import datetime
from typing import Any, Awaitable, Callable, Literal, Mapping, cast
from typing import Any, Literal, cast
from .context import build_call_args, describe_injection
from .errors import TaskFailedError, TaskTimeoutError
@@ -666,7 +667,7 @@ class AsyncLayerRunner:
return task_ctx, result
results = await asyncio.gather(*[_run_async_task(name) for name in to_run])
for name, (task_ctx, result) in zip(to_run, results):
for name, (task_ctx, result) in zip(to_run, results, strict=True):
_store_result(name, result, graph.resolved_spec(name), task_ctx, context, report, backend, on_event)
src/pyflowx/graph.py
+7 -13
View File
@@ -1,7 +1,7 @@
"""DAG 构建、校验、分层与可视化。
使用标准库的 :mod:`graphlib`（3.9+）或 :mod:`graphlib_backport`（3.8）
进行拓扑排序。图以增量方式构建并即时校验，使配置错误在构建时（而非执行时）快速失败。
使用标准库的 :mod:`graphlib` 进行拓扑排序。图以增量方式构建并即时校验，
使配置错误在构建时（而非执行时）快速失败。
支持：
* 图级默认值 :class:`GraphDefaults`，TaskSpec 字段为 ``None`` 时回退。
@@ -17,22 +17,16 @@ __all__ = [
"GraphDefaults",
]
import graphlib
import inspect
import sys
from collections.abc import Callable, Iterable, Mapping, Sequence
from dataclasses import dataclass, field, replace
from typing import Any, Callable, Iterable, Mapping, Sequence
from typing import Any
from .errors import CycleError, DuplicateTaskError, MissingDependencyError
from .task import Context, RetryPolicy, TaskSpec
if sys.version_info >= (3, 9): # pragma: no cover
import graphlib # pyright: ignore[reportUnreachable]
_TopologicalSorter = graphlib.TopologicalSorter
else: # pragma: no cover
import graphlib # type: ignore[import-untyped]
_TopologicalSorter = graphlib.TopologicalSorter # pragma: no cover
_TopologicalSorter = graphlib.TopologicalSorter
@dataclass
@@ -293,7 +287,7 @@ class Graph:
sorter = _TopologicalSorter(self.deps)
try:
sorter.prepare()
except graphlib.CycleError as exc: # type: ignore[name-defined]
except graphlib.CycleError as exc:
cycle: Sequence[str] = exc.args[1] if len(exc.args) > 1 else []
raise CycleError(list(cycle)) from exc
src/pyflowx/ops/_common.py
+1 -1
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from typing import Sequence
from collections.abc import Sequence
from pyflowx.conditions import Constants
src/pyflowx/ops/pymake.py
+14 -2
View File
@@ -221,11 +221,23 @@ def tc(cwd: Path = Path()) -> None:
"""类型检查 (聚合)."""
@px.tool("pymake", subcommand="p", help="推送代码 (清理 + push + push tags)", needs=["c", "git_push", "git_push_tags"], strategy="thread")
@px.tool(
"pymake",
subcommand="p",
help="推送代码 (清理 + push + push tags)",
needs=["c", "git_push", "git_push_tags"],
strategy="thread",
)
def p(cwd: Path = Path()) -> None:
"""推送代码 (聚合)."""
@px.tool("pymake", subcommand="pb", help="发布到 PyPI (twine + hatch)", needs=["twine_publish", "publish_python"], strategy="thread")
@px.tool(
"pymake",
subcommand="pb",
help="发布到 PyPI (twine + hatch)",
needs=["twine_publish", "publish_python"],
strategy="thread",
)
def pb(cwd: Path = Path()) -> None:
"""发布到 PyPI (聚合)."""
src/pyflowx/report.py
+2 -1
View File
@@ -6,8 +6,9 @@
from __future__ import annotations
from collections.abc import Iterator
from dataclasses import dataclass, field
from typing import Any, Iterator
from typing import Any
from .task import TaskResult, TaskStatus
src/pyflowx/runner.py
+2 -1
View File
@@ -14,9 +14,10 @@ from __future__ import annotations
import argparse
import enum
import sys
from collections.abc import Sequence
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Sequence, get_args
from typing import Any, get_args
from .compose import GraphComposer
from .errors import PyFlowXError
src/pyflowx/storage.py
+4 -4
View File
@@ -17,10 +17,10 @@ import json
import sys
import time
from abc import ABC, abstractmethod
from collections.abc import Iterator
from contextlib import contextmanager, nullcontext
from collections.abc import Iterator, Mapping
from contextlib import AbstractContextManager, contextmanager, nullcontext
from pathlib import Path
from typing import Any, ContextManager, Mapping
from typing import Any
if sys.version_info >= (3, 12):
from typing import override
@@ -63,7 +63,7 @@ class StateBackend(ABC):
:class:`JSONBackend` 在 :meth:`batch` 期间会延迟落盘，需在退出时调用。
"""
def batch(self) -> ContextManager[None]:
def batch(self) -> AbstractContextManager[None]:
"""返回一个上下文管理器，期间 :meth:`save` 可延迟 :meth:`flush`。
默认实现为 no-op（如 :class:`MemoryBackend`）。:class:`JSONBackend`
src/pyflowx/task.py
+6 -19
View File
@@ -22,7 +22,8 @@ import os
import shutil
import sys
import threading
from contextlib import contextmanager
from collections.abc import Callable, Coroutine, Generator, Mapping
from contextlib import AbstractContextManager, contextmanager
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
@@ -30,14 +31,7 @@ from functools import cached_property
from pathlib import Path
from typing import (
Any,
Callable,
ContextManager,
Coroutine,
Generator,
Generic,
List,
Mapping,
Union,
cast,
)
@@ -49,25 +43,18 @@ else:
T = TypeVar("T", default=Any)
# 任务可调用对象可以是同步或异步的。显式保留联合类型，让 mypy 理解两种形态。
TaskFn = Union[
Callable[..., T],
Callable[..., Coroutine[Any, Any, T]],
]
TaskFn = Callable[..., T] | Callable[..., Coroutine[Any, Any, T]]
# 跨任务结果映射。值刻意使用 ``Any``，因为不同任务返回不同类型；
# 单任务类型由函数签名本身保留。
Context = Mapping[str, Any]
# 命令类型支持
TaskCmd = Union[
List[str], # 命令列表, 如 ["ls", "-la"]
str, # shell 命令字符串
Callable[..., Any], # Python 函数
]
TaskCmd = list[str] | str | Callable[..., Any]
# 执行策略：sequential/thread/async 为层屏障模型，dependency 为依赖驱动模型。
Strategy = Union[str, "StrategyKind"]
StrategyKind = Any # 占位，避免循环；executors 模块用 Literal 约束
Strategy = str | StrategyKind
logger = logging.getLogger(__name__)
@@ -376,7 +363,7 @@ class TaskSpec(Generic[T]):
return shutil.which(cmd[0]) is not None
return True
def env_context(self) -> ContextManager[None]:
def env_context(self) -> AbstractContextManager[None]:
"""返回临时应用 ``env`` 与 ``cwd`` 的上下文管理器。
对 ``fn`` 任务生效。``cmd`` 任务在 :func:`_run_command` 中直接
src/pyflowx/tools.py
+206 -233
View File
@@ -1,12 +1,11 @@
"""@px.tool 装饰器: 函数签名驱动 CLI 生成 + DAG 编排.
"""@px.tool 装饰器: typer 驱动 CLI + DAG 编排.
替代 YAML 配置模式, 用 .py 装饰器统一描述工具.
函数签名 → argparse 自动生成, 函数体即任务逻辑, needs/strategy 表达 DAG.
函数签名 → typer 自动生成 CLI, 函数体即任务逻辑, needs/strategy 表达 DAG.
示例
----
::
@px.tool("pdftool", subcommand="m", help="合并 PDF")
def pdf_merge(
inputs: list[Path], # positional, nargs="+"
@@ -43,15 +42,23 @@ __all__ = [
"tool",
]
import argparse
import ast
import enum
import inspect
import sys
import textwrap
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Literal, Mapping, Sequence, Union, cast, get_args, get_origin, get_type_hints
from typing import Any, Literal, cast
import typer
from rich.console import Console
from rich.table import Table
# typer 内置 click 异常 (typer._click 与公共 click 是两套独立类层次)
from typer._click.exceptions import ClickException as _TyperClickException
from typer._click.exceptions import NoArgsIsHelpError as _NoArgsIsHelpError
from pyflowx.errors import PyFlowXError
from pyflowx.executors import run
@@ -219,170 +226,6 @@ def clear_tool_registry() -> None:
_TOOL_REGISTRY.clear()
# ---------------------------------------------------------------------- #
# CLI 生成 (函数签名 → argparse)
# ---------------------------------------------------------------------- #
def _snake_to_kebab(name: str) -> str:
"""snake_case → kebab-case."""
return name.replace("_", "-")
def _unwrap_optional(annotation: Any) -> tuple[Any, bool]:
"""返回 (实际类型, 是否 Optional).
``Optional[X]`` / ``X | None`` → ``(X, True)``; 其他 → ``(annotation, False)``.
"""
origin = get_origin(annotation)
is_union = origin is Union
if not is_union and sys.version_info >= (3, 10):
import types as _types
is_union = origin is _types.UnionType
if is_union:
args = [a for a in get_args(annotation) if a is not type(None)]
if len(args) == 1:
return args[0], True
return annotation, False
def _map_param_type(annotation: Any) -> dict[str, Any]:
"""将类型注解转为 argparse add_argument 的关键字参数.
Returns
-------
dict
含 ``type`` / ``nargs`` / ``choices`` 等; bool 不在此处理 (由调用方特殊处理)
"""
kwargs: dict[str, Any] = {}
inner, _is_opt = _unwrap_optional(annotation)
origin = get_origin(inner)
# list[X] / List[X] → nargs="+"
if origin is list:
args = get_args(inner)
if args:
kwargs["type"] = args[0]
kwargs["nargs"] = "+"
return kwargs
# Literal["a", "b"] → choices
if origin is Literal:
kwargs["choices"] = list(get_args(inner))
return kwargs
# 基本类型
if inner is int:
kwargs["type"] = int
elif inner is float:
kwargs["type"] = float
elif inner is Path:
kwargs["type"] = Path
elif inner is str:
kwargs["type"] = str
return kwargs
def _add_func_args_to_parser(parser: argparse.ArgumentParser, func: Callable[..., Any]) -> None:
"""从函数签名向 parser 添加位置参数和选项参数.
- 无默认值 → positional
- 有默认值 → ``--kebab-case`` 选项
- ``bool`` 默认 False → ``--flag`` store_true; 默认 True → ``--no-flag`` store_false
- ``list[X]`` → ``nargs="+"``
- ``Literal[...]`` → ``choices``
"""
sig = inspect.signature(func)
try:
hints: dict[str, Any] = get_type_hints(func)
except Exception:
hints = {}
for pname, param in sig.parameters.items():
if pname in ("self",):
continue
if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
continue
annotation = hints.get(pname, param.annotation if param.annotation is not inspect.Parameter.empty else str)
has_default = param.default is not inspect.Parameter.empty
default = param.default if has_default else inspect.Parameter.empty
# bool 特殊处理
if annotation is bool:
if not has_default or default is False:
parser.add_argument(
f"--{_snake_to_kebab(pname)}",
dest=pname,
action="store_true",
default=False,
help=pname,
)
else: # 默认 True → --no-flag store_false
parser.add_argument(
f"--no-{_snake_to_kebab(pname)}",
dest=pname,
action="store_false",
default=True,
help=f"禁用 {pname}",
)
continue
type_kwargs = _map_param_type(annotation)
if has_default:
flag = f"--{_snake_to_kebab(pname)}"
kwargs: dict[str, Any] = {"dest": pname, "default": default, "help": pname}
kwargs.update(type_kwargs)
parser.add_argument(flag, **kwargs)
else:
# 位置参数; 若类型是 list[X] 则 nargs="+" 已在 type_kwargs
kwargs = {"help": pname}
kwargs.update(type_kwargs)
parser.add_argument(pname, **kwargs)
def _add_global_options(parser: argparse.ArgumentParser) -> None:
"""向 parser (及所有 subparser) 添加全局选项."""
parser.add_argument("--dry-run", action="store_true", help="仅打印执行计划, 不执行")
parser.add_argument("--quiet", "-q", action="store_false", dest="verbose", help="减少输出")
parser.add_argument("--strategy", type=str, default=None, help="执行策略 (sequential/thread/dependency)")
parser.add_argument("--list", action="store_true", dest="list_jobs", help="列出所有任务后退出")
# 子命令也加全局选项
for action in parser._actions:
if action.dest == "command" and isinstance(action.choices, dict):
for sub in action.choices.values():
sub.add_argument("--dry-run", action="store_true")
sub.add_argument("--quiet", "-q", action="store_false", dest="verbose")
sub.add_argument("--strategy", type=str, default=None)
sub.add_argument("--list", action="store_true", dest="list_jobs")
def _build_parser(name: str, all_subs: dict[str | None, ToolSpec]) -> argparse.ArgumentParser:
"""构建 argparse 解析器.
单命令工具 (仅 subcommand=None) → 简单 parser;
多 subcommand 工具 → 带 subparsers.
"""
first_spec = next(iter(all_subs.values()))
description = first_spec.description or first_spec.help or name
parser = argparse.ArgumentParser(prog=f"pf {name}", description=description)
if None in all_subs and len(all_subs) == 1:
# 单命令工具
_add_func_args_to_parser(parser, all_subs[None].func)
else:
# 多 subcommand 工具
visible = [(sc, sp) for sc, sp in all_subs.items() if sc is not None and not sp.hidden]
if visible:
subparsers = parser.add_subparsers(dest="command", help="可用命令")
for sc, sp in sorted(visible, key=lambda x: x[0] or ""):
sub = subparsers.add_parser(sc, help=sp.help, description=sp.help)
_add_func_args_to_parser(sub, sp.func)
_add_global_options(parser)
return parser
# ---------------------------------------------------------------------- #
# 依赖收集 + TaskSpec 构建
# ---------------------------------------------------------------------- #
@@ -498,29 +341,178 @@ def _build_task_spec(spec: ToolSpec, variables: Mapping[str, Any]) -> TaskSpec:
)
def _extract_variables(args: argparse.Namespace, func: Callable[..., Any]) -> dict[str, Any]:
"""从 argparse namespace 提取函数签名对应的参数."""
sig = inspect.signature(func)
variables: dict[str, Any] = {}
for pname in sig.parameters:
if hasattr(args, pname):
variables[pname] = getattr(args, pname)
return variables
# ---------------------------------------------------------------------- #
# Typer app 构建 (函数签名 → typer 命令)
# ---------------------------------------------------------------------- #
def _global_option_params() -> list[inspect.Parameter]:
"""构建全局选项的 inspect.Parameter 列表.
各命令 dispatcher 追加这些参数到用户函数签名末尾, 使 ``--dry-run`` / ``--quiet`` /
``--strategy`` 出现在每个子命令上. ``--list`` 由 :func:`run_tool` 预处理.
"""
return [
inspect.Parameter(
"dry_run",
inspect.Parameter.KEYWORD_ONLY,
default=typer.Option(False, "--dry-run", help="仅打印执行计划, 不执行"),
annotation=bool,
),
inspect.Parameter(
"quiet",
inspect.Parameter.KEYWORD_ONLY,
default=typer.Option(False, "--quiet", "-q", help="减少输出"),
annotation=bool,
),
inspect.Parameter(
"strategy",
inspect.Parameter.KEYWORD_ONLY,
default=typer.Option(None, "--strategy", help="执行策略 (sequential/thread/dependency)"),
annotation=str | None,
),
]
def _make_dispatcher(tool_name: str, spec: ToolSpec) -> Callable[..., None]:
"""为 spec 创建 typer 命令 dispatcher.
dispatcher 签名 = 用户函数签名 (过滤 ``*args``/``**kwargs``) + 全局选项;
body 从 kwargs 分离全局选项后调用 :func:`_execute_dag`.
"""
func = spec.func
subcommand = spec.subcommand
user_sig = inspect.signature(func)
# 过滤 *args / **kwargs, 追加全局选项
params = [
p
for p in user_sig.parameters.values()
if p.kind not in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
]
params.extend(_global_option_params())
combined_sig = user_sig.replace(parameters=params)
def dispatcher(**kwargs: Any) -> None:
global_args = {
"dry_run": kwargs.pop("dry_run", False),
"verbose": not kwargs.pop("quiet", False),
"strategy": kwargs.pop("strategy", None),
}
_execute_dag(tool_name, subcommand, kwargs, global_args)
dispatcher.__signature__ = combined_sig # type: ignore[attr-defined]
dispatcher.__doc__ = spec.help or func.__doc__ or ""
dispatcher.__name__ = func.__name__
dispatcher.__annotations__ = {p.name: p.annotation for p in params}
return dispatcher
def _build_typer_app(name: str, all_subs: dict[str | None, ToolSpec]) -> typer.Typer:
"""构建 typer app.
单命令工具 (仅 subcommand=None) → callback(invoke_without_command=True);
多 subcommand 工具 → 各可见 subcommand 注册为命令 (``--list`` 由 run_tool 预处理).
"""
first_spec = next(iter(all_subs.values()))
description = first_spec.description or first_spec.help or name
is_single = None in all_subs and len(all_subs) == 1
app = typer.Typer(
name=f"pf {name}",
help=description,
# 单命令工具: 无参数时直接执行 callback; 多命令工具: 无 subcommand 时打印 help
no_args_is_help=not is_single,
add_completion=False,
rich_markup_mode="rich",
)
if is_single:
spec = all_subs[None]
dispatcher = _make_dispatcher(name, spec)
app.callback(invoke_without_command=True)(dispatcher)
return app
@app.callback()
def _callback() -> None:
"""工具入口; ``--list`` 由 :func:`run_tool` 预处理."""
for sc, spec in sorted(all_subs.items(), key=lambda x: x[0] or ""):
if sc is None or spec.hidden:
continue
dispatcher = _make_dispatcher(name, spec)
app.command(name=sc, help=spec.help)(dispatcher)
return app
# ---------------------------------------------------------------------- #
# 执行入口
# ---------------------------------------------------------------------- #
def _handle_list(name: str, all_subs: dict[str | None, ToolSpec]) -> int:
"""处理 --list, 打印任务列表."""
print(f"工具 {name} 的任务列表:")
def _execute_dag(
tool_name: str,
subcommand: str | None,
variables: dict[str, Any],
global_args: dict[str, Any],
) -> None:
"""收集依赖, 构建 Graph, 执行 DAG.
成功正常返回; 失败时打印 rich 错误并 ``raise typer.Exit``.
"""
all_subs = _TOOL_REGISTRY[tool_name]
if subcommand not in all_subs:
Console(stderr=True).print(f"[red]错误:[/red] 未知子命令 [yellow]{subcommand!r}[/yellow]")
raise typer.Exit(ToolExitCode.FAILURE)
target_spec = all_subs[subcommand]
# 收集依赖 + 构建子图
all_names = _collect_with_deps(tool_name, subcommand)
specs: list[TaskSpec] = []
for sc in all_names:
dep_spec = all_subs[sc]
specs.append(_build_task_spec(dep_spec, variables))
graph = Graph.from_specs(specs)
console = Console()
console.print(f"[bold cyan][{tool_name}][/bold cyan] 执行: [yellow]{subcommand or '全部任务'}[/yellow]")
try:
run(
graph,
strategy=global_args["strategy"] or target_spec.strategy or "dependency",
dry_run=global_args["dry_run"],
verbose=global_args["verbose"],
)
except PyFlowXError as e:
Console(stderr=True).print(f"[red]错误:[/red] {e}")
raise typer.Exit(ToolExitCode.FAILURE) from e
except KeyboardInterrupt:
Console(stderr=True).print("\n[yellow]已中断[/yellow]")
raise typer.Exit(ToolExitCode.INTERRUPTED) from None
def _handle_list_rich(name: str, all_subs: dict[str | None, ToolSpec]) -> None:
"""rich 表格打印任务列表."""
console = Console()
console.print(f"[bold]工具 {name} 的任务列表[/bold]")
table = Table(show_header=True, header_style="bold")
table.add_column("任务")
table.add_column("依赖")
table.add_column("类型")
table.add_column("状态")
for sc, spec in sorted(all_subs.items(), key=lambda x: (x[0] is not None, x[0] or "")):
display = sc if sc is not None else name
deps = ", ".join(spec.needs) if spec.needs else "(无依赖)"
hidden = " [隐藏]" if spec.hidden else ""
cmd_tag = " [cmd]" if spec.cmd is not None else ""
print(f" - {display} (依赖: {deps}){cmd_tag}{hidden}")
return 0
deps = ", ".join(spec.needs) if spec.needs else "—"
if spec.cmd is not None:
task_type = "cmd"
elif _is_aggregate(spec):
task_type = "聚合"
else:
task_type = "fn"
hidden = "[dim]隐藏[/dim]" if spec.hidden else ""
table.add_row(display, deps, task_type, hidden)
console.print(table)
def run_tool(name: str, argv: Sequence[str] | None = None) -> int:
@@ -539,61 +531,42 @@ def run_tool(name: str, argv: Sequence[str] | None = None) -> int:
退出码
"""
if name not in _TOOL_REGISTRY:
print(f"错误: 未注册工具 {name!r}", file=sys.stderr)
Console(stderr=True).print(f"[red]错误:[/red] 未注册工具 [yellow]{name!r}[/yellow]")
return ToolExitCode.FAILURE
all_subs = _TOOL_REGISTRY[name]
argv = list(argv) if argv is not None else sys.argv[1:]
parser = _build_parser(name, all_subs)
args = parser.parse_args(argv)
# --list 预处理: 出现在任意位置即列出任务后退出
if "--list" in argv:
_handle_list_rich(name, all_subs)
return ToolExitCode.SUCCESS
# --list 优先
if getattr(args, "list_jobs", False):
return _handle_list(name, all_subs)
# 确定要执行的 subcommand
visible_subs = [sc for sc, sp in all_subs.items() if sc is not None and not sp.hidden]
is_multi = len(visible_subs) > 1
if is_multi:
subcommand = getattr(args, "command", None)
if subcommand is None:
parser.print_help()
return ToolExitCode.SUCCESS
else:
# 单命令工具: subcommand 是 None 或唯一可见的具名 subcommand
subcommand = visible_subs[0] if visible_subs else None
if subcommand not in all_subs:
print(f"错误: 未知子命令 {subcommand!r}", file=sys.stderr)
return ToolExitCode.FAILURE
target_spec = all_subs[subcommand]
variables = _extract_variables(args, target_spec.func)
# 收集依赖 + 构建子图
all_names = _collect_with_deps(name, subcommand)
specs: list[TaskSpec] = []
for sc in all_names:
dep_spec = all_subs[sc]
specs.append(_build_task_spec(dep_spec, variables))
graph = Graph.from_specs(specs)
print(f"[{name}] 执行: {subcommand or '全部任务'}", flush=True)
app = _build_typer_app(name, all_subs)
try:
run(
graph,
strategy=target_spec.strategy or "dependency",
dry_run=getattr(args, "dry_run", False),
verbose=getattr(args, "verbose", True),
)
# standalone_mode=False: click 捕获 typer.Exit 后返回 exit_code (而非抛出);
# ClickException/Abort 则向上抛出, 由下方 except 分支处理
result = app(args=argv, standalone_mode=False)
# 正常完成返回 None; typer.Exit 返回 int exit_code
if isinstance(result, int) and result != 0:
return (
ToolExitCode(result)
if result in (ToolExitCode.SUCCESS, ToolExitCode.FAILURE, ToolExitCode.INTERRUPTED)
else ToolExitCode.FAILURE
)
return ToolExitCode.SUCCESS
except PyFlowXError as e:
print(f"错误: {e}", file=sys.stderr)
return ToolExitCode.FAILURE
except KeyboardInterrupt:
print("\n已中断", file=sys.stderr)
except _NoArgsIsHelpError:
# 多命令工具无 subcommand → 打印 help, 视为成功
return ToolExitCode.SUCCESS
except _TyperClickException as e:
return ToolExitCode.SUCCESS if e.exit_code == 0 else ToolExitCode.FAILURE
except typer.Abort:
return ToolExitCode.INTERRUPTED
except SystemExit as e:
code = e.code
if isinstance(code, int):
return code
return ToolExitCode.SUCCESS if code in (None, "0") else ToolExitCode.FAILURE
except KeyboardInterrupt:
return ToolExitCode.INTERRUPTED
tests/cli/test_emlmanager.py
+5 -3
View File
@@ -884,9 +884,11 @@ Date: Mon, {i + 1} Jan 2024 12:00:00 +0000
Body {i}
""")
with patch("sys.argv", ["emlmanager", "--dir", str(tmp_path), "--port", "8080"]), patch.object(
emlmanager, "ThreadingHTTPServer"
) as mock_server, patch("threading.Thread"):
with (
patch("sys.argv", ["emlmanager", "--dir", str(tmp_path), "--port", "8080"]),
patch.object(emlmanager, "ThreadingHTTPServer") as mock_server,
patch("threading.Thread"),
):
# Don't actually start the server
mock_server_instance = Mock()
mock_server.return_value = mock_server_instance
tests/cli/test_packtool.py
+15 -9
View File
@@ -103,9 +103,11 @@ class TestInstallEmbedPython:
output_dir = tmp_path / "python"
# Create a mock cache file that doesn't exist (force download)
with patch("platform.machine", return_value="x86_64"), patch(
"urllib.request.urlretrieve"
) as mock_urlretrieve, patch("zipfile.ZipFile") as mock_zipfile:
with (
patch("platform.machine", return_value="x86_64"),
patch("urllib.request.urlretrieve") as mock_urlretrieve,
patch("zipfile.ZipFile") as mock_zipfile,
):
# Mock successful download
mock_urlretrieve.return_value = None
mock_zip_instance = MagicMock()
@@ -190,9 +192,11 @@ class TestInstallEmbedPython:
"""Should handle different Python versions."""
output_dir = tmp_path / "python"
with patch("platform.machine", return_value="x86_64"), patch(
"urllib.request.urlretrieve"
) as mock_urlretrieve, patch("zipfile.ZipFile") as mock_zipfile:
with (
patch("platform.machine", return_value="x86_64"),
patch("urllib.request.urlretrieve") as mock_urlretrieve,
patch("zipfile.ZipFile") as mock_zipfile,
):
mock_zip_instance = MagicMock()
mock_zipfile.return_value.__enter__.return_value = mock_zip_instance
@@ -205,9 +209,11 @@ class TestInstallEmbedPython:
"""Should create cache directory and file."""
output_dir = tmp_path / "python"
with patch("platform.machine", return_value="x86_64"), patch(
"urllib.request.urlretrieve"
) as mock_urlretrieve, patch("zipfile.ZipFile") as mock_zipfile:
with (
patch("platform.machine", return_value="x86_64"),
patch("urllib.request.urlretrieve") as mock_urlretrieve,
patch("zipfile.ZipFile") as mock_zipfile,
):
mock_urlretrieve.return_value = None
mock_zip_instance = MagicMock()
mock_zipfile.return_value.__enter__.return_value = mock_zip_instance
tests/cli/test_pdftool.py
+5 -3
View File
@@ -247,9 +247,11 @@ class TestPdfOcr:
input_file.write_bytes(b"PDF content")
output_file = tmp_path / "ocr.pdf"
with patch("fitz.open") as mock_fitz_open, patch("PIL.Image.frombytes"), patch(
"pytesseract.image_to_string"
) as mock_ocr:
with (
patch("fitz.open") as mock_fitz_open,
patch("PIL.Image.frombytes"),
patch("pytesseract.image_to_string") as mock_ocr,
):
mock_doc = MagicMock()
mock_page = MagicMock()
mock_page.rect = MagicMock(width=800, height=600)
tests/cli/test_piptool.py
+4 -3
View File
@@ -121,9 +121,10 @@ class TestPipUninstall:
def test_pip_uninstall_with_wildcard(self) -> None:
"""Should handle wildcard in package name."""
with patch.object(dev, "_expand_wildcard_packages", return_value=["numpy", "numpy-core"]), patch(
"subprocess.run"
) as mock_run:
with (
patch.object(dev, "_expand_wildcard_packages", return_value=["numpy", "numpy-core"]),
patch("subprocess.run") as mock_run,
):
mock_run.return_value = MagicMock(returncode=0)
dev.pip_uninstall(["numpy*"])
assert mock_run.called
tests/cli/test_screenshot.py
+48 -24
View File
@@ -37,36 +37,48 @@ class TestTakeScreenshotFull:
def test_take_screenshot_full_windows(self, tmp_path: Path) -> None:
"""Should take full screenshot on Windows."""
with patch.object(Constants, "IS_WINDOWS", True), patch.object(Constants, "IS_MACOS", False), patch.object(
Path, "home", return_value=tmp_path
), patch("subprocess.run") as mock_run:
with (
patch.object(Constants, "IS_WINDOWS", True),
patch.object(Constants, "IS_MACOS", False),
patch.object(Path, "home", return_value=tmp_path),
patch("subprocess.run") as mock_run,
):
mock_run.return_value = MagicMock(returncode=0)
media.take_screenshot_full()
assert mock_run.called
def test_take_screenshot_full_macos(self, tmp_path: Path) -> None:
"""Should take full screenshot on macOS."""
with patch.object(Constants, "IS_WINDOWS", False), patch.object(Constants, "IS_MACOS", True), patch.object(
Path, "home", return_value=tmp_path
), patch("subprocess.run") as mock_run:
with (
patch.object(Constants, "IS_WINDOWS", False),
patch.object(Constants, "IS_MACOS", True),
patch.object(Path, "home", return_value=tmp_path),
patch("subprocess.run") as mock_run,
):
mock_run.return_value = MagicMock(returncode=0)
media.take_screenshot_full()
assert mock_run.called
def test_take_screenshot_full_linux(self, tmp_path: Path) -> None:
"""Should take full screenshot on Linux."""
with patch.object(Constants, "IS_WINDOWS", False), patch.object(Constants, "IS_MACOS", False), patch.object(
Path, "home", return_value=tmp_path
), patch("subprocess.run") as mock_run:
with (
patch.object(Constants, "IS_WINDOWS", False),
patch.object(Constants, "IS_MACOS", False),
patch.object(Path, "home", return_value=tmp_path),
patch("subprocess.run") as mock_run,
):
mock_run.return_value = MagicMock(returncode=0)
media.take_screenshot_full()
assert mock_run.called
def test_take_screenshot_full_linux_scrot_fallback(self, tmp_path: Path) -> None:
"""Should fallback to scrot when gnome-screenshot not found."""
with patch.object(Constants, "IS_WINDOWS", False), patch.object(Constants, "IS_MACOS", False), patch.object(
Path, "home", return_value=tmp_path
), patch("subprocess.run") as mock_run:
with (
patch.object(Constants, "IS_WINDOWS", False),
patch.object(Constants, "IS_MACOS", False),
patch.object(Path, "home", return_value=tmp_path),
patch("subprocess.run") as mock_run,
):
mock_run.side_effect = [FileNotFoundError(), MagicMock(returncode=0)]
media.take_screenshot_full()
assert mock_run.call_count == 2
@@ -80,36 +92,48 @@ class TestTakeScreenshotArea:
def test_take_screenshot_area_windows(self, tmp_path: Path) -> None:
"""Should take area screenshot on Windows."""
with patch.object(Constants, "IS_WINDOWS", True), patch.object(Constants, "IS_MACOS", False), patch.object(
Path, "home", return_value=tmp_path
), patch("subprocess.run") as mock_run:
with (
patch.object(Constants, "IS_WINDOWS", True),
patch.object(Constants, "IS_MACOS", False),
patch.object(Path, "home", return_value=tmp_path),
patch("subprocess.run") as mock_run,
):
mock_run.return_value = MagicMock(returncode=0)
media.take_screenshot_area()
assert mock_run.called
def test_take_screenshot_area_macos(self, tmp_path: Path) -> None:
"""Should take area screenshot on macOS."""
with patch.object(Constants, "IS_WINDOWS", False), patch.object(Constants, "IS_MACOS", True), patch.object(
Path, "home", return_value=tmp_path
), patch("subprocess.run") as mock_run:
with (
patch.object(Constants, "IS_WINDOWS", False),
patch.object(Constants, "IS_MACOS", True),
patch.object(Path, "home", return_value=tmp_path),
patch("subprocess.run") as mock_run,
):
mock_run.return_value = MagicMock(returncode=0)
media.take_screenshot_area()
assert mock_run.called
def test_take_screenshot_area_linux(self, tmp_path: Path) -> None:
"""Should take area screenshot on Linux."""
with patch.object(Constants, "IS_WINDOWS", False), patch.object(Constants, "IS_MACOS", False), patch.object(
Path, "home", return_value=tmp_path
), patch("subprocess.run") as mock_run:
with (
patch.object(Constants, "IS_WINDOWS", False),
patch.object(Constants, "IS_MACOS", False),
patch.object(Path, "home", return_value=tmp_path),
patch("subprocess.run") as mock_run,
):
mock_run.return_value = MagicMock(returncode=0)
media.take_screenshot_area()
assert mock_run.called
def test_take_screenshot_area_linux_scrot_fallback(self, tmp_path: Path) -> None:
"""Should fallback to scrot -s when gnome-screenshot not found."""
with patch.object(Constants, "IS_WINDOWS", False), patch.object(Constants, "IS_MACOS", False), patch.object(
Path, "home", return_value=tmp_path
), patch("subprocess.run") as mock_run:
with (
patch.object(Constants, "IS_WINDOWS", False),
patch.object(Constants, "IS_MACOS", False),
patch.object(Path, "home", return_value=tmp_path),
patch("subprocess.run") as mock_run,
):
mock_run.side_effect = [FileNotFoundError(), MagicMock(returncode=0)]
media.take_screenshot_area()
assert mock_run.call_count == 2
tests/cli/test_sshcopyid.py
+15 -9
View File
@@ -27,9 +27,11 @@ class TestSshCopyId:
pub_key = tmp_path / "id_rsa.pub"
pub_key.write_text("ssh-rsa AAAAB3...")
with patch.object(Path, "expanduser", return_value=pub_key), patch(
"subprocess.run", side_effect=FileNotFoundError
), pytest.raises(SystemExit):
with (
patch.object(Path, "expanduser", return_value=pub_key),
patch("subprocess.run", side_effect=FileNotFoundError),
pytest.raises(SystemExit),
):
system.ssh_copy_id("localhost", "user", "password")
def test_ssh_copy_id_timeout(self, tmp_path: Path) -> None:
@@ -37,9 +39,11 @@ class TestSshCopyId:
pub_key = tmp_path / "id_rsa.pub"
pub_key.write_text("ssh-rsa AAAAB3...")
with patch.object(Path, "expanduser", return_value=pub_key), patch(
"subprocess.run", side_effect=subprocess.TimeoutExpired("cmd", 30)
), pytest.raises(SystemExit):
with (
patch.object(Path, "expanduser", return_value=pub_key),
patch("subprocess.run", side_effect=subprocess.TimeoutExpired("cmd", 30)),
pytest.raises(SystemExit),
):
system.ssh_copy_id("localhost", "user", "password")
def test_ssh_copy_id_process_error(self, tmp_path: Path) -> None:
@@ -47,9 +51,11 @@ class TestSshCopyId:
pub_key = tmp_path / "id_rsa.pub"
pub_key.write_text("ssh-rsa AAAAB3...")
with patch.object(Path, "expanduser", return_value=pub_key), patch(
"subprocess.run", side_effect=subprocess.CalledProcessError(1, "cmd")
), pytest.raises(SystemExit):
with (
patch.object(Path, "expanduser", return_value=pub_key),
patch("subprocess.run", side_effect=subprocess.CalledProcessError(1, "cmd")),
pytest.raises(SystemExit),
):
system.ssh_copy_id("localhost", "user", "password")
def test_ssh_copy_id_success(self, tmp_path: Path) -> None:
tests/test_executors_edge_cases.py
+1 -1
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
import asyncio
import logging
import sys
from typing import Callable
from collections.abc import Callable
import pytest
tests/test_streaming.py
+1 -1
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from typing import Iterator
from collections.abc import Iterator
import pyflowx as px
tests/test_task_decorator.py
+2 -1
View File
@@ -2,8 +2,9 @@
from __future__ import annotations
from collections.abc import Mapping
from pathlib import Path
from typing import Any, Mapping
from typing import Any
import pyflowx as px
from pyflowx.task import RetryPolicy, TaskHooks, TaskSpec
tests/test_task_edge_cases.py
+13 -9
View File
@@ -151,9 +151,10 @@ def test_taskspec_list_cmd_timeout_mocked():
spec = TaskSpec("test", cmd=["sleep", "10"], timeout=0.1)
wrapped_fn = spec.effective_fn
with patch(
"subprocess.run", side_effect=subprocess.TimeoutExpired(cmd=["sleep", "10"], timeout=0.1)
), pytest.raises(RuntimeError, match="命令执行超时"):
with (
patch("subprocess.run", side_effect=subprocess.TimeoutExpired(cmd=["sleep", "10"], timeout=0.1)),
pytest.raises(RuntimeError, match="命令执行超时"),
):
_ = wrapped_fn()
@@ -162,8 +163,9 @@ def test_taskspec_shell_cmd_timeout_mocked():
spec = TaskSpec("test", cmd="sleep 10", timeout=0.1)
wrapped_fn = spec.effective_fn
with patch("subprocess.run", side_effect=subprocess.TimeoutExpired(cmd="sleep 10", timeout=0.1)), pytest.raises(
RuntimeError, match="Shell 命令执行超时"
with (
patch("subprocess.run", side_effect=subprocess.TimeoutExpired(cmd="sleep 10", timeout=0.1)),
pytest.raises(RuntimeError, match="Shell 命令执行超时"),
):
_ = wrapped_fn()
@@ -173,8 +175,9 @@ def test_taskspec_shell_cmd_file_not_found_mocked():
spec = TaskSpec("test", cmd="nonexistent_shell_command")
wrapped_fn = spec.effective_fn
with patch("subprocess.run", side_effect=FileNotFoundError("not found")), pytest.raises(
RuntimeError, match="Shell 命令未找到"
with (
patch("subprocess.run", side_effect=FileNotFoundError("not found")),
pytest.raises(RuntimeError, match="Shell 命令未找到"),
):
_ = wrapped_fn()
@@ -209,8 +212,9 @@ def test_taskspec_shell_cmd_os_error_mocked():
spec = TaskSpec("test", cmd="ls")
wrapped_fn = spec.effective_fn
with patch("subprocess.run", side_effect=OSError("os error")), pytest.raises(
RuntimeError, match="Shell 命令执行异常"
with (
patch("subprocess.run", side_effect=OSError("os error")),
pytest.raises(RuntimeError, match="Shell 命令执行异常"),
):
_ = wrapped_fn()
tests/test_tools.py
+245 -412
View File
@@ -1,14 +1,13 @@
"""tests/test_tools.py - @px.tool 装饰器核心模块测试.
覆盖 tools.py 全部分支:注册/重复检测/CLI 生成各类型/单命令执行/
覆盖 tools.py 全部分支:注册/重复检测/typer CLI 生成/单命令执行/
cmd/fn/聚合/DAG/hidden/全局选项/退出码三态.
"""
from __future__ import annotations
import argparse
from pathlib import Path
from typing import Any, Literal, Optional, Union
from typing import Any
import pytest
@@ -17,19 +16,15 @@ from pyflowx.executors import RunReport
from pyflowx.task import RetryPolicy
from pyflowx.tools import (
ToolExitCode,
_add_func_args_to_parser,
_add_global_options,
_build_parser,
_build_task_spec,
_build_typer_app,
_collect_with_deps,
_extract_variables,
_handle_list,
_global_option_params,
_handle_list_rich,
_has_function_logic,
_is_aggregate,
_map_param_type,
_make_dispatcher,
_noop,
_snake_to_kebab,
_unwrap_optional,
clear_tool_registry,
get_tool,
list_subcommands,
@@ -296,350 +291,6 @@ class TestRegistryErrors:
assert list_tools() == ["atool", "mtool", "ztool"]
# ---------------------------------------------------------------------- #
# _snake_to_kebab
# ---------------------------------------------------------------------- #
class TestSnakeToKebab:
"""测试 snake_case → kebab-case 转换."""
def test_simple(self) -> None:
assert _snake_to_kebab("output") == "output"
def test_snake(self) -> None:
assert _snake_to_kebab("output_dir") == "output-dir"
def test_multi_word(self) -> None:
assert _snake_to_kebab("python_mirror") == "python-mirror"
# ---------------------------------------------------------------------- #
# _unwrap_optional
# ---------------------------------------------------------------------- #
class TestUnwrapOptional:
"""测试 Optional 解包."""
def test_non_optional(self) -> None:
inner, is_opt = _unwrap_optional(int)
assert inner is int
assert is_opt is False
def test_optional(self) -> None:
inner, is_opt = _unwrap_optional(Optional[int])
assert inner is int
assert is_opt is True
def test_union_none(self) -> None:
inner, is_opt = _unwrap_optional(Union[int, None])
assert inner is int
assert is_opt is True
def test_str(self) -> None:
inner, is_opt = _unwrap_optional(str)
assert inner is str
assert is_opt is False
# ---------------------------------------------------------------------- #
# _map_param_type
# ---------------------------------------------------------------------- #
class TestMapParamType:
"""测试类型注解 → argparse kwargs 映射."""
def test_int(self) -> None:
assert _map_param_type(int) == {"type": int}
def test_float(self) -> None:
assert _map_param_type(float) == {"type": float}
def test_str(self) -> None:
assert _map_param_type(str) == {"type": str}
def test_path(self) -> None:
assert _map_param_type(Path) == {"type": Path}
def test_list_of_path(self) -> None:
kwargs = _map_param_type(list[Path])
assert kwargs["type"] is Path
assert kwargs["nargs"] == "+"
def test_list_no_type_arg(self) -> None:
"""list 无类型参数 → nargs='+', 无 type."""
kwargs = _map_param_type(list)
# list 的 origin 是 list,但 get_args(list) 为空
assert kwargs.get("nargs") == "+" or kwargs == {}
def test_literal(self) -> None:
kwargs = _map_param_type(Literal["a", "b"]) # type: ignore[valid-type]
assert kwargs["choices"] == ["a", "b"]
def test_unknown_type(self) -> None:
"""未知类型返回空 dict."""
assert _map_param_type(complex) == {}
def test_optional_int(self) -> None:
"""Optional[int] 解包后映射 int."""
kwargs = _map_param_type(Optional[int])
assert kwargs == {"type": int}
# ---------------------------------------------------------------------- #
# _add_func_args_to_parser
# ---------------------------------------------------------------------- #
class TestAddFuncArgsToParser:
"""测试函数签名 → argparse 参数."""
def _parse(self, func: Any, argv: list[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser()
_add_func_args_to_parser(parser, func)
return parser.parse_args(argv)
def test_positional(self) -> None:
def f(name: str) -> None:
pass
ns = self._parse(f, ["hello"])
assert ns.name == "hello"
def test_option_with_default(self) -> None:
def f(output: str = "out") -> None:
pass
ns = self._parse(f, ["--output", "x"])
assert ns.output == "x"
def test_option_default_value(self) -> None:
def f(output: str = "out") -> None:
pass
ns = self._parse(f, [])
assert ns.output == "out"
def test_int_type(self) -> None:
def f(level: int = 5) -> None:
pass
ns = self._parse(f, ["--level", "10"])
assert ns.level == 10
def test_float_type(self) -> None:
def f(rate: float = 1.0) -> None:
pass
ns = self._parse(f, ["--rate", "2.5"])
assert ns.rate == 2.5
def test_path_type(self) -> None:
def f(p: Path = Path()) -> None:
pass
ns = self._parse(f, ["--p", "/tmp"])
assert ns.p == Path("/tmp")
def test_list_positional_nargs_plus(self) -> None:
def f(inputs: list[Path]) -> None:
pass
ns = self._parse(f, ["a.pdf", "b.pdf"])
assert ns.inputs == [Path("a.pdf"), Path("b.pdf")]
def test_literal_choices(self) -> None:
def f(mode: Literal["a", "b"] = "a") -> None: # type: ignore[valid-type]
pass
ns = self._parse(f, ["--mode", "b"])
assert ns.mode == "b"
def test_bool_default_false_store_true(self) -> None:
def f(dry_run: bool = False) -> None:
pass
ns = self._parse(f, ["--dry-run"])
assert ns.dry_run is True
def test_bool_default_false_absent(self) -> None:
def f(dry_run: bool = False) -> None:
pass
ns = self._parse(f, [])
assert ns.dry_run is False
def test_bool_default_true_store_false(self) -> None:
def f(verbose: bool = True) -> None:
pass
ns = self._parse(f, ["--no-verbose"])
assert ns.verbose is False
def test_bool_default_true_absent(self) -> None:
def f(verbose: bool = True) -> None:
pass
ns = self._parse(f, [])
assert ns.verbose is True
def test_bool_no_default_store_true(self) -> None:
"""bool 无默认值 → store_true, 默认 False."""
def f(flag: bool) -> None:
pass
ns = self._parse(f, ["--flag"])
assert ns.flag is True
def test_skip_self(self) -> None:
"""self 参数被跳过."""
class C:
def m(self, x: int = 1) -> None:
pass
ns = self._parse(C().m, ["--x", "5"])
assert ns.x == 5
def test_skip_var_positional(self) -> None:
"""*args 被跳过."""
def f(*args: Any, x: int = 1) -> None:
pass
ns = self._parse(f, ["--x", "5"])
assert ns.x == 5
def test_skip_var_keyword(self) -> None:
"""**kwargs 被跳过."""
def f(x: int = 1, **kw: Any) -> None:
pass
ns = self._parse(f, ["--x", "5"])
assert ns.x == 5
def test_kebab_case_option(self) -> None:
"""snake_case 参数名 → kebab-case 选项."""
def f(output_dir: str = "out") -> None:
pass
ns = self._parse(f, ["--output-dir", "x"])
assert ns.output_dir == "x"
def test_no_annotation_defaults_to_str(self) -> None:
"""无类型注解 → str."""
def f(x="default") -> None: # type: ignore[no-untyped-def]
pass
ns = self._parse(f, ["--x", "val"])
assert ns.x == "val"
# ---------------------------------------------------------------------- #
# _add_global_options
# ---------------------------------------------------------------------- #
class TestAddGlobalOptions:
"""测试全局选项注入."""
def test_global_options_on_simple_parser(self) -> None:
parser = argparse.ArgumentParser()
_add_global_options(parser)
ns = parser.parse_args(["--dry-run", "--quiet", "--strategy", "thread", "--list"])
assert ns.dry_run is True
assert ns.verbose is False
assert ns.strategy == "thread"
assert ns.list_jobs is True
def test_global_options_on_subparsers(self) -> None:
"""subparser 也获得全局选项."""
def f(x: int = 1) -> None:
pass
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(dest="command")
sub = subparsers.add_parser("cmd")
_add_func_args_to_parser(sub, f)
_add_global_options(parser)
ns = parser.parse_args(["cmd", "--x", "5", "--dry-run", "--quiet"])
assert ns.x == 5
assert ns.dry_run is True
assert ns.verbose is False
# ---------------------------------------------------------------------- #
# _build_parser
# ---------------------------------------------------------------------- #
class TestBuildParser:
"""测试 parser 构建."""
def test_single_command_parser(self) -> None:
"""单命令工具(仅 subcommand=None)→ 简单 parser."""
@tool("solo")
def f(x: int = 1) -> None:
pass
from pyflowx.tools import _TOOL_REGISTRY
parser = _build_parser("solo", _TOOL_REGISTRY["solo"])
ns = parser.parse_args(["--x", "5"])
assert ns.x == 5
def test_multi_subcommand_parser(self) -> None:
"""多 subcommand → 带 subparsers."""
@tool("t", subcommand="a")
def a(x: int = 1) -> None:
pass
@tool("t", subcommand="b")
def b(y: str = "z") -> None:
pass
from pyflowx.tools import _TOOL_REGISTRY
parser = _build_parser("t", _TOOL_REGISTRY["t"])
ns = parser.parse_args(["a", "--x", "5"])
assert ns.command == "a"
assert ns.x == 5
def test_parser_description_from_description(self) -> None:
"""description 优先于 help."""
@tool("t", description="工具描述")
def f() -> None:
pass
from pyflowx.tools import _TOOL_REGISTRY
parser = _build_parser("t", _TOOL_REGISTRY["t"])
assert parser.description == "工具描述"
def test_parser_description_from_help(self) -> None:
"""无 description 时用 help."""
@tool("t", help="帮助文本")
def f() -> None:
pass
from pyflowx.tools import _TOOL_REGISTRY
parser = _build_parser("t", _TOOL_REGISTRY["t"])
assert parser.description == "帮助文本"
def test_parser_description_from_name(self) -> None:
"""无 description/help 时用工具名."""
@tool("t")
def f() -> None:
pass
from pyflowx.tools import _TOOL_REGISTRY
parser = _build_parser("t", _TOOL_REGISTRY["t"])
assert parser.description == "t"
# ---------------------------------------------------------------------- #
# _collect_with_deps
# ---------------------------------------------------------------------- #
@@ -963,38 +614,178 @@ class TestBuildTaskSpec:
# ---------------------------------------------------------------------- #
# _extract_variables
# _global_option_params
# ---------------------------------------------------------------------- #
class TestExtractVariables:
"""测试 argparse namespace → 函数参数提取."""
class TestGlobalOptionParams:
"""测试全局选项参数构建."""
def test_extract_matching_params(self) -> None:
def f(x: int = 1, y: str = "z") -> None:
def test_count(self) -> None:
"""返回 3 个参数 (dry_run, quiet, strategy)."""
params = _global_option_params()
assert len(params) == 3
def test_names(self) -> None:
"""参数名正确."""
params = _global_option_params()
assert [p.name for p in params] == ["dry_run", "quiet", "strategy"]
def test_all_keyword_only(self) -> None:
"""所有参数为 KEYWORD_ONLY."""
import inspect
params = _global_option_params()
assert all(p.kind == inspect.Parameter.KEYWORD_ONLY for p in params)
def test_annotations(self) -> None:
"""注解正确."""
params = _global_option_params()
assert params[0].annotation is bool # dry_run
assert params[1].annotation is bool # quiet
# strategy: str | None
assert params[2].annotation is not None
# ---------------------------------------------------------------------- #
# _make_dispatcher
# ---------------------------------------------------------------------- #
class TestMakeDispatcher:
"""测试 typer 命令 dispatcher 构建."""
def test_signature_includes_user_params(self) -> None:
"""dispatcher 签名包含用户函数参数."""
@tool("t", subcommand="a")
def a(x: int = 1, y: str = "z") -> None:
pass
ns = argparse.Namespace(x=5, y="a", extra="ignored")
variables = _extract_variables(ns, f)
assert variables == {"x": 5, "y": "a"}
import inspect
def test_extract_missing_attr(self) -> None:
"""namespace 缺少属性 → 不包含在结果中."""
spec = get_tool("t", "a")
dispatcher = _make_dispatcher("t", spec)
sig = inspect.signature(dispatcher)
assert "x" in sig.parameters
assert "y" in sig.parameters
def f(x: int = 1, y: str = "z") -> None:
def test_signature_includes_global_options(self) -> None:
"""dispatcher 签名包含全局选项."""
@tool("t", subcommand="a")
def a(x: int = 1) -> None:
pass
ns = argparse.Namespace(x=5)
variables = _extract_variables(ns, f)
assert variables == {"x": 5}
import inspect
spec = get_tool("t", "a")
dispatcher = _make_dispatcher("t", spec)
sig = inspect.signature(dispatcher)
assert "dry_run" in sig.parameters
assert "quiet" in sig.parameters
assert "strategy" in sig.parameters
def test_signature_filters_var_positional(self) -> None:
"""*args 被过滤."""
@tool("t", subcommand="a")
def a(*args: Any, x: int = 1) -> None:
pass
import inspect
spec = get_tool("t", "a")
dispatcher = _make_dispatcher("t", spec)
sig = inspect.signature(dispatcher)
assert "args" not in sig.parameters
assert "x" in sig.parameters
def test_signature_filters_var_keyword(self) -> None:
"""**kwargs 被过滤."""
@tool("t", subcommand="a")
def a(x: int = 1, **kw: Any) -> None:
pass
import inspect
spec = get_tool("t", "a")
dispatcher = _make_dispatcher("t", spec)
sig = inspect.signature(dispatcher)
assert "kw" not in sig.parameters
def test_dispatcher_name_and_doc(self) -> None:
"""dispatcher 继承函数名和 docstring."""
@tool("t", subcommand="a", help="帮助文本")
def a(x: int = 1) -> None:
pass
spec = get_tool("t", "a")
dispatcher = _make_dispatcher("t", spec)
assert dispatcher.__name__ == "a"
assert dispatcher.__doc__ == "帮助文本"
# ---------------------------------------------------------------------- #
# _handle_list
# _build_typer_app
# ---------------------------------------------------------------------- #
class TestHandleList:
"""测试 --list 输出."""
class TestBuildTyperApp:
"""测试 typer app 构建."""
def test_single_command_app(self) -> None:
"""单命令工具 → typer app 有 callback."""
@tool("solo")
def f(x: int = 1) -> None:
pass
from pyflowx.tools import _TOOL_REGISTRY
app = _build_typer_app("solo", _TOOL_REGISTRY["solo"])
assert app is not None
def test_multi_subcommand_app(self) -> None:
"""多 subcommand 工具 → typer app 有 commands."""
@tool("t", subcommand="a")
def a(x: int = 1) -> None:
pass
@tool("t", subcommand="b")
def b(y: str = "z") -> None:
pass
from pyflowx.tools import _TOOL_REGISTRY
app = _build_typer_app("t", _TOOL_REGISTRY["t"])
assert app is not None
def test_hidden_not_registered_as_command(self) -> None:
"""hidden subcommand 不注册为 typer 命令."""
@tool("t", subcommand="visible")
def visible() -> None:
pass
@tool("t", subcommand="hidden", hidden=True)
def hidden() -> None:
pass
from pyflowx.tools import _TOOL_REGISTRY
app = _build_typer_app("t", _TOOL_REGISTRY["t"])
# typer app 的 registered_commands 不应包含 hidden
command_names = [cmd.name for cmd in app.registered_commands]
assert "visible" in command_names
assert "hidden" not in command_names
# ---------------------------------------------------------------------- #
# _handle_list_rich
# ---------------------------------------------------------------------- #
class TestHandleListRich:
"""测试 rich 表格任务列表输出."""
def test_list_output(self, capsys: pytest.CaptureFixture[str]) -> None:
"""--list 打印任务列表."""
"""--list 打印 rich 任务列表."""
@tool("t", subcommand="a", needs=["b"])
def a() -> None:
@@ -1010,28 +801,44 @@ class TestHandleList:
from pyflowx.tools import _TOOL_REGISTRY
ret = _handle_list("t", _TOOL_REGISTRY["t"])
assert ret == 0
_handle_list_rich("t", _TOOL_REGISTRY["t"])
out = capsys.readouterr().out
assert "工具 t 的任务列表:" in out
assert "工具 t 的任务列表" in out
assert "a" in out
assert "b" in out
assert "h" in out # hidden 也在 list 中显示
assert "[cmd]" in out # b 是 cmd 任务
assert "[隐藏]" in out # h 是 hidden
assert "cmd" in out # b 是 cmd 任务
assert "隐藏" in out # h 是 hidden
def test_list_no_deps(self, capsys: pytest.CaptureFixture[str]) -> None:
"""无依赖显示 (无依赖)."""
def test_list_aggregate_type(self, capsys: pytest.CaptureFixture[str]) -> None:
"""聚合任务显示 '聚合' 类型."""
@tool("t", subcommand="a")
def a() -> None:
@tool("t", subcommand="agg", needs=["dep"])
def agg() -> None:
pass
@tool("t", subcommand="dep")
def dep() -> None:
pass
from pyflowx.tools import _TOOL_REGISTRY
_handle_list("t", _TOOL_REGISTRY["t"])
_handle_list_rich("t", _TOOL_REGISTRY["t"])
out = capsys.readouterr().out
assert "(无依赖)" in out
assert "聚合" in out
def test_list_fn_type(self, capsys: pytest.CaptureFixture[str]) -> None:
"""fn 任务显示 'fn' 类型."""
@tool("t", subcommand="f")
def f() -> None:
print("hi")
from pyflowx.tools import _TOOL_REGISTRY
_handle_list_rich("t", _TOOL_REGISTRY["t"])
out = capsys.readouterr().out
assert "fn" in out
# ---------------------------------------------------------------------- #
@@ -1047,7 +854,7 @@ class TestRunTool:
err = capsys.readouterr().err
assert "未注册工具" in err
def test_single_command_success(self, mock_run: list[Any], capsys: pytest.CaptureFixture[str]) -> None:
def test_single_command_success(self, mock_run: list[Any]) -> None:
"""单命令工具执行成功 → 0."""
@tool("solo")
@@ -1098,23 +905,40 @@ class TestRunTool:
assert kwargs["verbose"] is False
def test_strategy_override(self, mock_run: list[Any]) -> None:
"""--strategy 透传到 run."""
"""--strategy CLI 选项覆盖 spec.strategy."""
@tool("t", strategy="thread")
def f() -> None:
pass
run_tool("t", ["--strategy", "sequential"])
_, kwargs = mock_run[0]
assert kwargs["strategy"] == "sequential"
def test_strategy_from_spec(self, mock_run: list[Any]) -> None:
"""无 --strategy 时用 spec.strategy."""
# 注意: run_tool 用 target_spec.strategy or "dependency"
# 单命令无 strategy → "dependency"
@tool("t", strategy="thread")
def f() -> None:
pass
run_tool("t", [])
_, kwargs = mock_run[0]
# strategy 来自 target_spec.strategy
# run_tool 调用 run(graph, strategy=target_spec.strategy or "dependency", ...)
# mock_run 捕获 kwargs
assert kwargs["strategy"] == "thread" or "strategy" in kwargs
assert kwargs["strategy"] == "thread"
def test_strategy_default(self, mock_run: list[Any]) -> None:
"""无 --strategy 无 spec.strategy → 'dependency'."""
@tool("t")
def f() -> None:
pass
run_tool("t", [])
_, kwargs = mock_run[0]
assert kwargs["strategy"] == "dependency"
def test_list_flag(self, capsys: pytest.CaptureFixture[str]) -> None:
"""--list → 打印任务列表 + 退出 0."""
"""--list → 打印 rich 任务列表 + 退出 0."""
@tool("t", subcommand="a")
def a() -> None:
@@ -1123,10 +947,22 @@ class TestRunTool:
ret = run_tool("t", ["--list"])
assert ret == ToolExitCode.SUCCESS
out = capsys.readouterr().out
assert "工具 t 的任务列表:" in out
assert "工具 t 的任务列表" in out
def test_list_flag_with_subcommand(self, capsys: pytest.CaptureFixture[str]) -> None:
"""--list 与子命令同时出现 → 仍列出任务."""
@tool("t", subcommand="a")
def a() -> None:
pass
ret = run_tool("t", ["a", "--list"])
assert ret == ToolExitCode.SUCCESS
out = capsys.readouterr().out
assert "工具 t 的任务列表" in out
def test_multi_no_subcommand_prints_help(self, capsys: pytest.CaptureFixture[str]) -> None:
"""多 subcommand 工具无 subcommand → 打印 help + 0."""
"""多 subcommand 工具无 subcommand → typer 打印 help."""
@tool("t", subcommand="a")
def a() -> None:
@@ -1137,10 +973,8 @@ class TestRunTool:
pass
ret = run_tool("t", [])
# typer no_args_is_help=True → SystemExit(0)
assert ret == ToolExitCode.SUCCESS
out = capsys.readouterr().out
# 应打印 help( argparse 的 print_help 输出包含 usage)
assert "usage" in out.lower() or "pf t" in out
def test_task_failed_error(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
"""TaskFailedError(继承 PyFlowXError) → 退出码 1 + stderr."""
@@ -1197,6 +1031,16 @@ class TestRunTool:
err = capsys.readouterr().err
assert "已中断" in err
def test_unknown_option_returns_failure(self) -> None:
"""未知选项 → ClickException → 退出码 1."""
@tool("t")
def f() -> None:
pass
ret = run_tool("t", ["--nonexistent-option"])
assert ret == ToolExitCode.FAILURE
def test_run_with_deps(self, mock_run: list[Any]) -> None:
"""有依赖时构建子图(多个 TaskSpec)."""
@@ -1250,32 +1094,21 @@ class TestIntegration:
ret = run_tool("pymake", ["ba", "--cwd", "/tmp"])
assert ret == ToolExitCode.SUCCESS
# 验证图中包含 3 个任务
_graph, _ = mock_run[0]
# cwd 应从 CLI 透传(但 _build_task_spec 对 cmd 任务从 variables['cwd'] 取)
# variables 包含 cwd=/tmp
# b 和 bc 的 cwd 应为 /tmp
# ba 是聚合,无 cwd
def test_full_pdftool_like_flow(self, mock_run: list[Any]) -> None:
"""模拟 pdftool:m 合并 PDF."""
@tool("pdftool", subcommand="m", help="合并 PDF")
def merge(
inputs: list[Path],
output: Path = Path("merged.pdf"),
input_paths: list[Path],
output_path: Path = Path("merged.pdf"),
password: str = "",
) -> None:
"""合并 PDF 文件."""
pass
ret = run_tool("pdftool", ["m", "a.pdf", "b.pdf", "--output", "out.pdf"])
ret = run_tool("pdftool", ["m", "a.pdf", "b.pdf", "--output-path", "out.pdf"])
assert ret == ToolExitCode.SUCCESS
# 验证 kwargs 透传
_graph, _ = mock_run[0]
# fn 任务的 kwargs 应包含 inputs/output/password
def test_list_tools_after_multiple_registrations(self) -> None:
"""多次注册后 list_tools 排序正确."""
tox.ini
+1 -1
View File
@@ -1,6 +1,6 @@
[tox]
isolated_build = true
envlist = py38, py39, py310, py311, py312, py313, py314
envlist = py310, py311, py312, py313, py314
min_version = 4.0
requires = tox-uv
skipsdist = true
typings/graphlib/graphlib.pyi
+2 -1
View File
@@ -2,7 +2,8 @@
This type stub file was generated by pyright.
"""
from typing import Any, Generator
from collections.abc import Generator
from typing import Any
__all__ = ["CycleError", "TopologicalSorter"]
_NODE_OUT = ...
uv.lock
Generated
+206 -4028
View File
File diff suppressed because it is too large Load Diff