"""YAML 任务编排加载器.
参考 GitHub Actions 风格, 从 YAML 文件构建任务图.
支持 jobs/needs/strategy.matrix/if 等核心 CI/CD 概念, 实现声明式串并行任务编排.
Schema 概览
-----------
::
strategy: thread # 图级默认策略
defaults: # 图级默认值
retry: {max_attempts: 3}
verbose: true
jobs:
setup: # job_id 即任务名
run: "git clone https://github.com/foo/bar"
build:
needs: [setup] # 依赖列表
cmd: ["python", "-m", "build"]
strategy: # 矩阵扇出
matrix:
python: ["3.8", "3.9"]
if: "success()" # 条件
timeout: 300
retry: {max_attempts: 2, delay: 1.0}
cwd: ./build
env: {DEBUG: "1"}
verbose: true
continue-on-error: false
tags: [ci]
runs-on: linux
priority: 10
concurrency-key: deploy_lock
deploy:
needs: [build]
run: "twine upload dist/*"
if: "env.DEPLOY_TOKEN != ''"
allow-upstream-skip: true
字段映射
-------
- ``cmd`` / ``run``: ``cmd`` 为命令列表, ``run`` 为 shell 字符串
- ``fn``: 引用 ``register_fn`` 注册的 Python 函数名, 配合 ``args``/``kwargs`` 传参
- ``needs``: 对应 ``depends_on``
- ``if``: 条件表达式, 支持 ``success()`` / ``always()`` / ``env.VAR`` / ``env.VAR == 'x'``
- ``strategy.matrix``: 笛卡尔积展开为多个任务, 任务名追加 ``_{key}-{val}`` 后缀
- ``runs-on``: 追加到 ``tags``
- 其余字段名连字符转下划线后透传给 :class:`~pyflowx.task.TaskSpec`
矩阵占位符
----------
矩阵值通过 ``${{ matrix.key }}`` 占位符注入 ``cmd`` / ``run`` / ``cwd`` / ``env`` / ``args`` / ``kwargs`` 中::
jobs:
test:
cmd: ["python{{ matrix.version }}", "-m", "pytest"]
strategy:
matrix:
version: ["3.8", "3.9"]
变量占位符
----------
运行时变量通过 ``${VAR}`` 占位符注入, 变量由 ``load_yaml(path, variables={...})``
或 ``parse_yaml_string(content, variables={...})`` 传入::
# Python
graph = px.load_yaml("pipeline.yaml", variables={"PROJECT_DIR": "./project"})
# YAML
jobs:
pack:
fn: pack_source
args: ["${PROJECT_DIR}", "./dist"]
函数引用
----------
``fn`` 字段引用 ``@px.register_fn`` 注册的函数, 配合 ``args``/``kwargs`` 传参::
# Python
@px.register_fn("pack_source")
def pack_source(project_dir, output_dir):
...
# YAML
jobs:
pack:
fn: pack_source
args: ["./project", "./dist"]
needs: [clean]
"""
from __future__ import annotations
__all__ = ["YamlLoadError", "build_cli_parser", "load_yaml", "parse_yaml_string", "run_cli", "run_yaml"]
import itertools
import re
import sys
from pathlib import Path
from typing import TYPE_CHECKING, Any, Iterable, Mapping, Sequence, cast
import pyflowx as px
from pyflowx.conditions import BuiltinConditions
from pyflowx.executors import Strategy
from pyflowx.registry import get_fn as _get_fn
from pyflowx.task import Condition, RetryPolicy, TaskSpec
if TYPE_CHECKING:
from pyflowx.executors import EventCallback, RunReport
from pyflowx.storage import StateBackend
# 导入 ops 模块以注册 CLI 工具函数, 使 YAML fn 字段可引用
# 使用 try/except 避免最小安装场景下的导入失败
try:
from pyflowx.ops import dev as _dev # noqa: F401
from pyflowx.ops import files as _files # noqa: F401
from pyflowx.ops import media as _media # noqa: F401
from pyflowx.ops import system as _system # noqa: F401
except ImportError: # pragma: no cover
pass
# 矩阵占位符: ${{ matrix.key }}
_MATRIX_PLACEHOLDER = re.compile(r"\$\{\{\s*matrix\.(\w+)\s*\}\}")
# 变量占位符: ${VAR}
_VARIABLE_PLACEHOLDER = re.compile(r"\$\{(\w+)\}")
# 变量全匹配: 字符串完全等于 ${VAR} (用于返回变量实际值而非字符串化)
_VARIABLE_FULL_MATCH = re.compile(r"^\$\{(\w+)\}$")
# 条件表达式
_ENV_VAR_EXISTS_RE = re.compile(r"^env\.(\w+)$")
_ENV_VAR_EQUALS_RE = re.compile(r"^env\.(\w+)\s*==\s*['\"]([^'\"]*)['\"]$")
_ENV_VAR_NOT_EQUALS_RE = re.compile(r"^env\.(\w+)\s*!=\s*['\"]([^'\"]*)['\"]$")
# 支持的 job 字段名（连字符形式）
_JOB_FIELDS_HYPHEN = frozenset(
{
"continue-on-error",
"skip-if-missing",
"allow-upstream-skip",
"concurrency-key",
"runs-on",
}
)
class YamlLoadError(px.PyFlowXError):
"""YAML 加载错误."""
def _aggregate_noop() -> None:
"""聚合 job 的占位函数.
方向 B: 有 ``needs`` 无 ``cmd/run/fn`` 的 job 仅作为依赖聚合点,
执行时调用此 no-op 函数, 不产生副作用, 返回 ``None``.
"""
return None
[文档]
def load_yaml(path: str | Path, variables: Mapping[str, Any] | None = None) -> px.Graph:
"""从 YAML 文件加载任务图.
Parameters
----------
path : str | Path
YAML 文件路径
variables : Mapping[str, Any] | None
运行时变量, 用于替换 ``${VAR}`` 占位符
Returns
-------
px.Graph
构建好的任务图
Raises
------
YamlLoadError
文件不存在、YAML 格式错误、schema 校验失败、循环依赖等
"""
p = Path(path)
if not p.exists():
raise YamlLoadError(f"YAML 文件不存在: {p}")
try:
content = p.read_text(encoding="utf-8")
except OSError as e:
raise YamlLoadError(f"读取文件失败: {e}") from e
return parse_yaml_string(content, variables=variables)
[文档]
def parse_yaml_string(content: str, variables: Mapping[str, Any] | None = None) -> px.Graph:
"""从 YAML 字符串构建任务图.
Parameters
----------
content : str
YAML 文本
variables : Mapping[str, Any] | None
运行时变量, 用于替换 ``${VAR}`` 占位符
Returns
-------
px.Graph
构建好的任务图
"""
data = _parse_yaml(content)
if not isinstance(data, Mapping):
raise YamlLoadError("YAML 根节点必须是 mapping (对象)")
return _build_graph(data, variables=variables)
[文档]
def run_yaml(
path: str | Path,
jobs: str | Sequence[str] | None = None,
variables: Mapping[str, Any] | None = None,
strategy: str | None = None,
*,
max_workers: int | None = None,
dry_run: bool = False,
verbose: bool = False,
on_event: EventCallback | None = None,
state: StateBackend | None = None,
concurrency_limits: Mapping[str, int] | None = None,
) -> RunReport:
"""加载 YAML 配置并执行任务图.
便捷函数, 组合 :func:`load_yaml` + (可选) 子图选择 + :func:`pyflowx.run`.
指定 ``jobs`` 时自动包含其传递依赖, 适用于从多 job 配置中执行单个命令.
Parameters
----------
path : str | Path
YAML 配置文件路径
jobs : str | Sequence[str] | None
要执行的 job 名. None 表示执行全部. str 表示单个 job.
指定 job 时自动包含其传递依赖.
variables : Mapping[str, Any] | None
运行时变量, 用于替换 ``${VAR}`` 占位符
strategy : str | None
执行策略, None 使用 YAML 中的策略或默认 ``"dependency"``
max_workers : int | None
``"thread"`` 策略的线程池大小
dry_run : bool
仅打印执行计划, 不执行
verbose : bool
打印任务生命周期
on_event : EventCallback | None
状态转换回调
state : StateBackend | None
断点续跑后端
concurrency_limits : Mapping[str, int] | None
``{concurrency_key: max_concurrent}`` 映射
Returns
-------
RunReport
执行报告
Raises
------
YamlLoadError
YAML 加载或 schema 校验失败
KeyError
``jobs`` 指定了不存在的 job 名
"""
graph = load_yaml(path, variables=variables)
if jobs is not None:
job_names = [jobs] if isinstance(jobs, str) else list(jobs)
all_names = _collect_with_deps(graph, job_names)
graph = graph.subgraph_by_names(all_names)
effective_strategy = strategy if strategy is not None else (graph.defaults.strategy or "dependency")
return px.run(
graph,
strategy=cast(Strategy, effective_strategy),
max_workers=max_workers,
dry_run=dry_run,
verbose=verbose,
on_event=on_event,
state=state,
concurrency_limits=concurrency_limits,
)
def _collect_with_deps(graph: px.Graph, names: Iterable[str]) -> set[str]:
"""收集 ``names`` 及其所有传递依赖, 用于子图选择."""
result: set[str] = set()
queue: list[str] = list(names)
while queue:
name = queue.pop()
if name in result:
continue
result.add(name)
spec = graph.spec(name)
queue.extend(spec.depends_on)
return result
def _parse_yaml(content: str) -> Any:
"""解析 YAML 文本, 失败抛出 YamlLoadError."""
try:
import yaml
except ImportError as e: # pragma: no cover
raise YamlLoadError("缺少 PyYAML 依赖, 请执行 `pip install pyyaml`") from e
try:
return yaml.safe_load(content)
except yaml.YAMLError as e:
raise YamlLoadError(f"YAML 解析失败: {e}") from e
def _merge_variables(
yaml_vars: Any,
caller_vars: Mapping[str, Any] | None,
) -> dict[str, Any]:
"""合并 YAML 中的 variables (默认值) 与调用方传入的 variables (后者优先)."""
if isinstance(yaml_vars, Mapping):
return {**yaml_vars, **(caller_vars or {})}
return dict(caller_vars) if caller_vars else {}
def _build_graph(data: Mapping[str, Any], variables: Mapping[str, Any] | None = None) -> px.Graph:
"""从解析后的 mapping 构建 Graph."""
jobs = data.get("jobs")
if not isinstance(jobs, Mapping):
raise YamlLoadError("YAML 必须包含 jobs 字段, 且 jobs 必须是 mapping")
if not jobs:
raise YamlLoadError("jobs 不能为空")
merged_vars = _merge_variables(data.get("variables"), variables)
defaults = _parse_defaults(data.get("defaults") or {})
# 顶层 strategy 覆盖 defaults.strategy
top_strategy = data.get("strategy")
if top_strategy is not None:
if not isinstance(top_strategy, str):
raise YamlLoadError("顶层 strategy 必须是字符串")
defaults = px.GraphDefaults(
retry=defaults.retry,
timeout=defaults.timeout,
strategy=top_strategy,
tags=defaults.tags,
env=defaults.env,
cwd=defaults.cwd,
priority=defaults.priority,
continue_on_error=defaults.continue_on_error,
concurrency_key=defaults.concurrency_key,
verbose=defaults.verbose,
)
# 第一阶段: 预计算每个 job 的展开名 (用于矩阵依赖展开)
job_expansions: dict[str, list[str]] = {}
for job_id, job_config in jobs.items():
if not isinstance(job_id, str) or not job_id:
raise YamlLoadError(f"job id 必须是非空字符串, 收到: {job_id!r}")
if not isinstance(job_config, Mapping):
raise YamlLoadError(f"job {job_id!r} 必须是 mapping")
matrix_combos = _expand_matrix(job_config.get("strategy"))
if any(combo is not None for combo in matrix_combos):
job_expansions[job_id] = [_matrix_name(job_id, c) for c in matrix_combos]
else:
job_expansions[job_id] = [job_id]
# 第二阶段: 解析 job, 展开矩阵依赖
specs: list[TaskSpec[Any]] = []
seen_names: set[str] = set()
for job_id, job_config in jobs.items():
for spec in _parse_job(job_id, job_config, job_expansions, variables=merged_vars):
if spec.name in seen_names:
raise YamlLoadError(f"重复任务名: {spec.name!r}")
seen_names.add(spec.name)
specs.append(spec)
return px.Graph.from_specs(specs, defaults=defaults)
def _parse_defaults(data: Mapping[str, Any]) -> px.GraphDefaults:
"""解析 defaults 字段为 GraphDefaults."""
if not data:
return px.GraphDefaults()
retry = _parse_retry(data["retry"]) if "retry" in data else None
return px.GraphDefaults(
retry=retry,
timeout=_as_float(data.get("timeout")),
strategy=_as_str(data.get("strategy")),
tags=_as_str_tuple(data.get("tags")),
env=_as_str_mapping(data.get("env")),
cwd=_as_path(data.get("cwd")),
priority=_as_int(data.get("priority")) or 0,
continue_on_error=bool(data.get("continue-on-error") or data.get("continue_on_error") or False),
concurrency_key=_as_str(data.get("concurrency-key") or data.get("concurrency_key")),
verbose=bool(data.get("verbose") or False),
)
def _parse_job(
job_id: str,
config: Mapping[str, Any],
job_expansions: Mapping[str, list[str]],
variables: Mapping[str, Any] | None = None,
) -> list[TaskSpec[Any]]:
"""解析单个 job, 返回 spec 列表 (矩阵展开可能产生多个).
Parameters
----------
job_id : str
job 标识符
config : Mapping[str, Any]
job 配置
job_expansions : Mapping[str, list[str]]
每个 job_id 到其展开名列表的映射, 用于展开矩阵依赖
variables : Mapping[str, Any] | None
运行时变量, 用于替换 ``${VAR}`` 占位符
"""
cmd_data, run_data, fn_data, args_data, kwargs_data = _parse_task_type_fields(job_id, config)
matrix_combos = _expand_matrix(config.get("strategy"))
if not matrix_combos:
matrix_combos = [None]
needs_tuple = _expand_needs(_as_str_tuple(config.get("needs")), job_expansions)
if_data = config.get("if")
timeout = _as_float(config.get("timeout"))
retry = _parse_retry(config["retry"]) if "retry" in config else None
cwd = _as_path(config.get("cwd"))
env = _as_str_mapping(config.get("env"))
verbose = bool(config.get("verbose"))
continue_on_error = bool(config.get("continue-on-error") or config.get("continue_on_error") or False)
skip_if_missing = bool(config.get("skip-if-missing") or config.get("skip_if_missing") or False)
allow_upstream_skip = bool(config.get("allow-upstream-skip") or config.get("allow_upstream_skip") or False)
priority = _as_int(config.get("priority")) or 0
concurrency_key = _as_str(config.get("concurrency-key") or config.get("concurrency_key"))
tags = _as_str_tuple(config.get("tags"))
runs_on = _as_str(config.get("runs-on") or config.get("runs_on"))
if runs_on:
tags = (*tags, runs_on)
fn = _resolve_fn(job_id, fn_data)
# 方向 B: 聚合 job (有 needs 无 cmd/run/fn) 注入 no-op fn, 仅作依赖聚合点
is_aggregate = fn is None and cmd_data is None and run_data is None and bool(needs_tuple)
if is_aggregate:
fn = _aggregate_noop
specs: list[TaskSpec[Any]] = []
for combo in matrix_combos:
name = _matrix_name(job_id, combo)
cmd = _build_cmd(cmd_data, run_data, combo, variables) if fn_data is None and not is_aggregate else None
cwd_resolved = Path(_replace_in_str(str(cwd), combo, variables)) if cwd is not None else None
env_resolved = _replace_in_mapping(env, combo, variables) if env else None
conditions = _parse_condition(if_data) if if_data is not None else ()
# fn 任务的 args/kwargs 应用占位符替换
args_resolved = _replace_in_value(args_data, combo, variables) if args_data is not None else ()
kwargs_resolved = _replace_in_value(kwargs_data, combo, variables) if kwargs_data is not None else {}
spec = px.TaskSpec(
name=name,
fn=fn,
cmd=cmd,
args=tuple(args_resolved),
kwargs=dict(kwargs_resolved),
depends_on=needs_tuple,
retry=retry if retry is not None else RetryPolicy(),
timeout=timeout,
tags=tags,
conditions=conditions,
cwd=cwd_resolved,
env=env_resolved,
verbose=verbose,
skip_if_missing=skip_if_missing,
allow_upstream_skip=allow_upstream_skip,
priority=priority,
concurrency_key=concurrency_key,
continue_on_error=continue_on_error,
)
specs.append(spec)
return specs
def _parse_task_type_fields(
job_id: str,
config: Mapping[str, Any],
) -> tuple[Any, Any, Any, Any, Any]:
"""解析并校验任务类型字段 (cmd/run/fn) 及参数 (args/kwargs).
Returns
-------
tuple
(cmd_data, run_data, fn_data, args_data, kwargs_data)
"""
cmd_data = config.get("cmd")
run_data = config.get("run")
fn_data = config.get("fn")
args_data = config.get("args")
kwargs_data = config.get("kwargs")
# 校验互斥: cmd/run/fn 三选一
provided = [x for x in (cmd_data, run_data, fn_data) if x is not None]
if len(provided) > 1:
raise YamlLoadError(f"job {job_id!r} 不能同时提供 cmd/run/fn")
# 方向 B: 允许有 needs 无 cmd/run/fn 的聚合 job (仅作为依赖聚合点)
if not provided and not config.get("needs"):
raise YamlLoadError(f"job {job_id!r} 必须提供 cmd/run/fn 之一, 或提供 needs 作为聚合 job")
# fn 必须是字符串
if fn_data is not None and not isinstance(fn_data, str):
raise YamlLoadError(f"job {job_id!r} 的 fn 必须是字符串, 收到: {type(fn_data).__name__}")
# args 必须是列表, kwargs 必须是 mapping
if args_data is not None and not isinstance(args_data, list):
raise YamlLoadError(f"job {job_id!r} 的 args 必须是列表, 收到: {type(args_data).__name__}")
if kwargs_data is not None and not isinstance(kwargs_data, Mapping):
raise YamlLoadError(f"job {job_id!r} 的 kwargs 必须是 mapping, 收到: {type(kwargs_data).__name__}")
return cmd_data, run_data, fn_data, args_data, kwargs_data
def _expand_needs(
needs_raw: tuple[str, ...],
job_expansions: Mapping[str, list[str]],
) -> tuple[str, ...]:
"""展开 needs 中的矩阵依赖."""
needs: list[str] = []
for dep in needs_raw:
if dep in job_expansions:
needs.extend(job_expansions[dep])
else:
needs.append(dep)
return tuple(needs)
def _resolve_fn(job_id: str, fn_data: Any) -> Any:
"""从 registry 查找已注册的函数."""
if fn_data is None:
return None
try:
return _get_fn(fn_data)
except KeyError as e:
raise YamlLoadError(f"job {job_id!r} 引用的函数 {fn_data!r} 未注册") from e
def _expand_matrix(strategy_data: Any) -> list[Mapping[str, Any] | None]:
"""展开 strategy.matrix 为笛卡尔积列表.
Returns
-------
list[Mapping[str, Any] | None]
无矩阵时返回 [None], 有矩阵时返回笛卡尔积
"""
if strategy_data is None:
return [None]
if not isinstance(strategy_data, Mapping):
raise YamlLoadError("strategy 必须是 mapping")
matrix = strategy_data.get("matrix")
if matrix is None:
return [None]
if not isinstance(matrix, Mapping):
raise YamlLoadError("strategy.matrix 必须是 mapping")
if not matrix:
return [None]
keys = list(matrix.keys())
value_lists = []
for key in keys:
vals = matrix[key]
if isinstance(vals, (str, int, float, bool)):
vals = [vals]
elif not isinstance(vals, list):
raise YamlLoadError(f"matrix.{key} 必须是列表或标量")
if not vals:
raise YamlLoadError(f"matrix.{key} 不能为空列表")
value_lists.append([(key, v) for v in vals])
combos: list[Mapping[str, Any] | None] = []
for combo_tuples in itertools.product(*value_lists):
combos.append(dict(combo_tuples))
return combos
def _matrix_name(job_id: str, combo: Mapping[str, Any] | None) -> str:
"""生成矩阵展开后的任务名."""
if combo is None:
return job_id
parts = [job_id]
for key, val in combo.items():
parts.append(f"{key}-{_sanitize_matrix_value(val)}")
return "_".join(parts)
def _sanitize_matrix_value(val: Any) -> str:
"""将矩阵值转为合法任务名片段."""
s = str(val)
# 替换任务名中不允许的字符
return re.sub(r"[^a-zA-Z0-9._-]", "_", s)
def _build_cmd(
cmd_data: Any,
run_data: Any,
combo: Mapping[str, Any] | None,
variables: Mapping[str, Any] | None = None,
) -> list[str] | str:
"""构建 cmd 字段, 应用占位符替换.
列表元素若完全匹配 ``${VAR}`` 且变量为列表时, 展开为多个元素;
否则按字符串替换处理.
"""
if cmd_data is not None:
if isinstance(cmd_data, str):
return _replace_in_str(cmd_data, combo, variables)
if isinstance(cmd_data, list):
result: list[str] = []
for item in cmd_data:
if variables and isinstance(item, str):
m = _VARIABLE_FULL_MATCH.match(item)
if m and m.group(1) in variables:
val = variables[m.group(1)]
if isinstance(val, list):
result.extend(str(v) for v in val)
continue
result.append(_replace_in_str(str(item), combo, variables))
return result
raise YamlLoadError(f"cmd 必须是 list 或 str, 收到: {type(cmd_data).__name__}")
# run_data 不为 None (已在外层校验)
return _replace_in_str(str(run_data), combo, variables)
def _replace_in_str(
text: str,
combo: Mapping[str, Any] | None,
variables: Mapping[str, Any] | None = None,
) -> str:
"""替换字符串中的 ``${VAR}`` 变量占位符和 ``${{ matrix.key }}`` 矩阵占位符."""
# 先替换变量占位符 ${VAR}
if variables:
def _sub_var(match: re.Match[str]) -> str:
key = match.group(1)
if key not in variables:
raise YamlLoadError(f"变量 {key!r} 未定义")
return str(variables[key])
text = _VARIABLE_PLACEHOLDER.sub(_sub_var, text)
# 再替换矩阵占位符 ${{ matrix.key }}
if combo is not None:
def _sub_matrix(match: re.Match[str]) -> str:
key = match.group(1)
if key not in combo:
raise YamlLoadError(f"矩阵占位符 matrix.{key} 未定义")
return str(combo[key])
text = _MATRIX_PLACEHOLDER.sub(_sub_matrix, text)
return text
def _replace_in_mapping(
env: Mapping[str, str],
combo: Mapping[str, Any] | None,
variables: Mapping[str, Any] | None = None,
) -> dict[str, str]:
"""替换 env 映射中的占位符."""
return {k: _replace_in_str(v, combo, variables) for k, v in env.items()}
def _replace_in_value(
value: Any,
combo: Mapping[str, Any] | None,
variables: Mapping[str, Any] | None = None,
) -> Any:
"""递归替换值 (list/mapping/scalar) 中的占位符.
当字符串完全等于 ``${VAR}`` 时, 返回变量的实际值 (非字符串化),
适用于 ``args: ["${FILES}"]`` 这种需要传递列表/对象而非字符串的场景.
"""
if isinstance(value, str) and variables:
m = _VARIABLE_FULL_MATCH.match(value)
if m and m.group(1) in variables:
return variables[m.group(1)]
if isinstance(value, str):
return _replace_in_str(value, combo, variables)
if isinstance(value, list):
return [_replace_in_value(v, combo, variables) for v in value]
if isinstance(value, Mapping):
return {k: _replace_in_value(v, combo, variables) for k, v in value.items()}
return value
def _parse_condition(expr: Any) -> tuple[Condition, ...]:
"""解析 if 条件表达式为 Condition 元组."""
if not isinstance(expr, str):
raise YamlLoadError(f"if 条件必须是字符串, 收到: {type(expr).__name__}")
expr = expr.strip()
if not expr:
raise YamlLoadError("if 条件不能为空")
lowered = expr.lower()
if lowered in ("success()", "success"):
# 默认行为: 依赖成功才执行. 不设 condition.
return ()
if lowered in ("always()", "always"):
# always() 在 PyFlowX 中无直接对应, 返回空 condition 元组,
# 由调用方通过 allow_upstream_skip=True 实现.
return ()
if lowered in ("failure()", "failure"):
raise YamlLoadError("if: failure() 在 PyFlowX 中不支持 (失败会中止或跳过下游)")
# env.VAR_EXISTS
m = _ENV_VAR_EXISTS_RE.match(expr)
if m:
return (BuiltinConditions.ENV_VAR_EXISTS(m.group(1)),)
# env.VAR == 'value'
m = _ENV_VAR_EQUALS_RE.match(expr)
if m:
return (BuiltinConditions.ENV_VAR_EQUALS(m.group(1), m.group(2)),)
# env.VAR != 'value'
m = _ENV_VAR_NOT_EQUALS_RE.match(expr)
if m:
return (BuiltinConditions.NOT(BuiltinConditions.ENV_VAR_EQUALS(m.group(1), m.group(2))),)
raise YamlLoadError(f"不支持的 if 表达式: {expr!r} (支持 success()/always()/env.VAR/env.VAR=='x')")
def _parse_retry(data: Any) -> RetryPolicy:
"""解析 retry 配置为 RetryPolicy."""
if data is None:
return RetryPolicy()
if isinstance(data, Mapping):
return RetryPolicy(
max_attempts=int(data.get("max_attempts", data.get("max-attempts", 1))),
delay=float(data.get("delay", 0.0)),
backoff=float(data.get("backoff", 1.0)),
jitter=float(data.get("jitter", 0.0)),
retry_on=(Exception,), # 默认重试所有异常
)
raise YamlLoadError(f"retry 必须是 mapping, 收到: {type(data).__name__}")
def _as_float(value: Any) -> float | None:
"""安全转换为 float."""
if value is None:
return None
if isinstance(value, (int, float)):
return float(value)
raise YamlLoadError(f"期望数字, 收到: {value!r}")
def _as_int(value: Any) -> int | None:
"""安全转换为 int."""
if value is None:
return None
if isinstance(value, int) and not isinstance(value, bool):
return value
if isinstance(value, float):
return int(value)
raise YamlLoadError(f"期望整数, 收到: {value!r}")
def _as_str(value: Any) -> str | None:
"""安全转换为 str."""
if value is None:
return None
if isinstance(value, str):
return value
raise YamlLoadError(f"期望字符串, 收到: {value!r}")
def _as_str_tuple(value: Any) -> tuple[str, ...]:
"""转换为字符串元组."""
if value is None:
return ()
if isinstance(value, str):
return (value,)
if isinstance(value, list):
return tuple(str(v) for v in value)
raise YamlLoadError(f"期望字符串或列表, 收到: {value!r}")
def _as_str_mapping(value: Any) -> dict[str, str] | None:
"""转换为 {str: str} 映射."""
if value is None:
return None
if not isinstance(value, Mapping):
raise YamlLoadError(f"期望 mapping, 收到: {value!r}")
return {str(k): str(v) for k, v in value.items()}
def _as_path(value: Any) -> Path | None:
"""转换为 Path."""
if value is None:
return None
return Path(str(value))
# ============================================================================
# CLI 参数解析
# ============================================================================
# 支持的 CLI 参数类型
_CLI_TYPE_MAP: dict[str, type] = {
"str": str,
"int": int,
"float": float,
"path": Path,
}
def _parse_cli_type(type_str: str | None) -> type:
"""解析 CLI 参数类型字符串."""
if type_str is None:
return str
if type_str in _CLI_TYPE_MAP:
return _CLI_TYPE_MAP[type_str]
raise YamlLoadError(f"不支持的 CLI 参数类型: {type_str!r}")
[文档]
def build_cli_parser(data: Mapping[str, Any]) -> Any:
"""从 YAML 配置的 ``cli:`` 段构建 argparse 解析器.
Parameters
----------
data : Mapping[str, Any]
解析后的 YAML 配置 (包含 ``cli`` 和 ``jobs``)
Returns
-------
argparse.ArgumentParser
构建好的参数解析器
Raises
------
YamlLoadError
cli 段格式错误
"""
import argparse
cli_raw = data.get("cli")
if cli_raw is None:
cli_config: Mapping[str, Any] = {}
elif not isinstance(cli_raw, Mapping):
raise YamlLoadError(f"cli 必须是 mapping, 收到: {type(cli_raw).__name__}")
else:
cli_config = cli_raw
description = str(cli_config.get("description", ""))
usage = str(cli_config.get("usage", "")) if cli_config.get("usage") else None
parser = argparse.ArgumentParser(description=description, usage=usage)
subcommands = cli_config.get("subcommands")
if subcommands is not None:
if not isinstance(subcommands, Mapping):
raise YamlLoadError(f"cli.subcommands 必须是 mapping, 收到: {type(subcommands).__name__}")
subparsers = parser.add_subparsers(dest="command", help="可用命令")
for sub_name, sub_config in subcommands.items():
if not isinstance(sub_config, Mapping):
raise YamlLoadError(f"cli.subcommands.{sub_name} 必须是 mapping")
sub = subparsers.add_parser(sub_name, help=str(sub_config.get("help", "")))
_add_args_to_parser(sub, sub_config)
else:
_add_args_to_parser(parser, cli_config)
return parser
def _add_args_to_parser(parser: Any, config: Mapping[str, Any]) -> None: # noqa: PLR0912
"""向 parser 添加位置参数和选项参数."""
positional = config.get("positional", [])
if positional:
if not isinstance(positional, list):
raise YamlLoadError("positional 必须是列表")
for arg_cfg in positional:
if not isinstance(arg_cfg, Mapping):
raise YamlLoadError("positional 元素必须是 mapping")
name = str(arg_cfg["name"])
kwargs: dict[str, Any] = {}
if "help" in arg_cfg:
kwargs["help"] = str(arg_cfg["help"])
if "nargs" in arg_cfg:
kwargs["nargs"] = arg_cfg["nargs"]
if "type" in arg_cfg:
kwargs["type"] = _parse_cli_type(str(arg_cfg["type"]))
if "default" in arg_cfg:
kwargs["default"] = arg_cfg["default"]
parser.add_argument(name.lower().replace("-", "_"), **kwargs)
options = config.get("options", [])
if options:
if not isinstance(options, list):
raise YamlLoadError("options 必须是列表")
for opt_cfg in options:
if not isinstance(opt_cfg, Mapping):
raise YamlLoadError("options 元素必须是 mapping")
name = str(opt_cfg["name"])
flag = str(opt_cfg.get("flag", f"--{name.lower().replace('_', '-')}"))
kwargs = {}
if "help" in opt_cfg:
kwargs["help"] = str(opt_cfg["help"])
if "type" in opt_cfg:
kwargs["type"] = _parse_cli_type(str(opt_cfg["type"]))
if "default" in opt_cfg:
kwargs["default"] = opt_cfg["default"]
if "required" in opt_cfg:
kwargs["required"] = bool(opt_cfg["required"])
if "action" in opt_cfg:
kwargs["action"] = str(opt_cfg["action"])
parser.add_argument(flag, dest=name.lower().replace("-", "_"), **kwargs)
class YamlCliRunner:
"""从 YAML 配置构建并执行 CLI.
读取 YAML 的 ``cli:`` 段构建 argparse 解析器, 解析命令行参数,
然后调用 :func:`run_yaml` 执行对应的 job.
全局选项 ``--dry-run`` / ``--quiet`` / ``--strategy`` / ``--list``
自动添加到所有工具.
"""
def __init__(self, path: str | Path, argv: Sequence[str] | None = None) -> None:
self._path = Path(path)
self._argv = list(argv) if argv is not None else sys.argv[1:]
self._data: Mapping[str, Any] = {}
self._cli_config: Mapping[str, Any] = {}
def run(self) -> int: # noqa: PLR0911
"""执行 CLI, 返回退出码 (0 成功, 1 失败)."""
from pyflowx.errors import TaskFailedError
if not self._path.exists():
print(f"错误: YAML 文件不存在: {self._path}", file=sys.stderr)
return 1
if not self._load_config():
return 1
try:
parser = build_cli_parser(self._data)
except YamlLoadError as e:
print(f"错误: CLI 配置错误: {e}", file=sys.stderr)
return 1
self._add_global_options(parser)
args = parser.parse_args(self._argv)
has_subcommands = bool(self._cli_config.get("subcommands"))
cmd = getattr(args, "command", None) if has_subcommands else None
job_name: str | None = cmd
# --list 优先于 cmd 缺失判断
if getattr(args, "list_jobs", False):
return self._handle_list()
if has_subcommands and cmd is None:
parser.print_help()
return 0
variables = self._extract_variables(args)
job_display = job_name if job_name else "全部任务"
print(f"[{self._path.stem}] 执行: {job_display}", flush=True)
try:
run_yaml(
self._path,
jobs=job_name,
variables=variables if variables else None,
strategy=getattr(args, "strategy", None),
dry_run=getattr(args, "dry_run", False),
verbose=getattr(args, "verbose", False),
)
return 0
except TaskFailedError as e:
print(f"错误: {e}", file=sys.stderr)
return 1
except YamlLoadError as e:
print(f"错误: {e}", file=sys.stderr)
return 1
except KeyboardInterrupt:
print("\n已中断", file=sys.stderr)
return 130
def _load_config(self) -> bool:
"""加载并校验 YAML 配置, 返回是否成功."""
try:
content = self._path.read_text(encoding="utf-8")
data = _parse_yaml(content)
except YamlLoadError as e:
print(f"错误: YAML 加载失败: {e}", file=sys.stderr)
return False
if not isinstance(data, Mapping):
print("错误: YAML 根节点必须是 mapping", file=sys.stderr)
return False
cli_raw = data.get("cli") or {}
if not isinstance(cli_raw, Mapping):
print("错误: cli 段格式错误", file=sys.stderr)
return False
self._data = data
self._cli_config = cli_raw
return True
def _add_global_options(self, parser: Any) -> None:
"""向解析器添加全局选项.
verbose 默认开启 (显示执行过程), 用 --quiet 关闭.
"""
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 hasattr(action, "choices"):
for sub in action.choices.values():
sub.add_argument("--dry-run", action="store_true", help="仅打印执行计划, 不执行")
sub.add_argument("--quiet", "-q", action="store_false", dest="verbose", help="减少输出")
sub.add_argument("--strategy", type=str, default=None, help="执行策略")
sub.add_argument("--list", action="store_true", dest="list_jobs", help="列出所有任务名后退出")
def _extract_variables(self, args: Any) -> dict[str, Any]:
"""从解析后的 namespace 提取变量字典 (大写 key)."""
variables: dict[str, Any] = {}
arg_names: set[str] = set()
subcommands = self._cli_config.get("subcommands")
if subcommands is not None and isinstance(subcommands, Mapping):
cmd = getattr(args, "command", None)
if cmd and cmd in subcommands:
sub_cfg = subcommands[cmd]
if isinstance(sub_cfg, Mapping):
self._collect_arg_names(arg_names, sub_cfg)
else:
self._collect_arg_names(arg_names, self._cli_config)
for name in arg_names:
attr_name = name.lower().replace("-", "_")
if hasattr(args, attr_name):
variables[name] = getattr(args, attr_name)
return variables
@staticmethod
def _collect_arg_names(names: set[str], config: Mapping[str, Any]) -> None:
"""收集配置中的参数名."""
for arg_cfg in config.get("positional", []) or []:
if isinstance(arg_cfg, Mapping) and "name" in arg_cfg:
names.add(str(arg_cfg["name"]))
for opt_cfg in config.get("options", []) or []:
if isinstance(opt_cfg, Mapping) and "name" in opt_cfg:
names.add(str(opt_cfg["name"]))
def _handle_list(self) -> int:
"""处理 --list 选项, 打印任务列表."""
graph = load_yaml(self._path)
print("任务列表:")
for name in graph.names:
spec = graph.spec(name)
deps = ", ".join(spec.depends_on) if spec.depends_on else "(无依赖)"
print(f" - {name} (依赖: {deps})")
return 0
[文档]
def run_cli(path: str | Path, argv: Sequence[str] | None = None) -> int:
"""从 YAML 配置构建 CLI 并执行.
:class:`YamlCliRunner` 的便捷包装.
Parameters
----------
path : str | Path
YAML 配置文件路径
argv : Sequence[str] | None
命令行参数, None 表示使用 sys.argv[1:]
Returns
-------
int
退出码 (0 成功, 1 失败)
"""
return YamlCliRunner(path, argv).run()