API 参考

任务描述

class pyflowx.TaskSpec(name: str, fn: ~typing.Callable[[...], ~pyflowx.task.T] | ~typing.Callable[[...], ~typing.Coroutine[~typing.Any, ~typing.Any, ~pyflowx.task.T]] | None = None, cmd: ~typing.List[str] | str | ~typing.Callable[[...], ~typing.Any] | None = None, depends_on: tuple[str, ...] = (), soft_depends_on: tuple[str, ...] = (), defaults: ~typing.Mapping[str, ~typing.Any] = <factory>, args: tuple[~typing.Any, ...] = (), kwargs: ~typing.Mapping[str, ~typing.Any] = <factory>, retry: ~pyflowx.task.RetryPolicy = <factory>, timeout: float | None = None, tags: tuple[str, ...] = (), conditions: tuple[~typing.Callable[[~typing.Mapping[str, ~typing.Any]], bool], ...] = (), cwd: ~pathlib.Path | None = None, env: ~typing.Mapping[str, str] | None = None, verbose: bool = False, skip_if_missing: bool = False, allow_upstream_skip: bool = False, strategy: str | None = None, priority: int = 0, concurrency_key: str | None = None, continue_on_error: bool = False, cache_key: ~typing.Callable[[~typing.Mapping[str, ~typing.Any]], str] | None = None, hooks: ~pyflowx.task.TaskHooks = <factory>, executor: str = 'thread')[源代码]

基类：Generic[T]

单个 DAG 节点的不可变描述。

参数

name:: 任务在图内的唯一标识。其他任务通过 depends_on 引用此名称。
fn:: 待执行的可调用对象，可为同步或异步。其参数名驱动自动上下文注入（见 pyflowx.context）。若提供 cmd 参数，则此参数会被忽略。
cmd:: 命令列表或 shell 字符串，支持三种形态： - list[str]: 命令及参数列表，如 ["ls", "-la"] - str: shell 命令字符串，如 "pip freeze > requirements.txt" - Callable: Python 函数，与 fn 参数等效
depends_on:: 硬依赖任务名。必须全部成功完成才会运行本任务。上游被 SKIPPED 时，本任务也会被 SKIPPED（除非 allow_upstream_skip=True）。
soft_depends_on:: 软依赖任务名。会等待其完成，但其结果不影响本任务是否执行： - 上游成功：注入其返回值 - 上游 SKIPPED 或失败：注入 defaults 中提供的默认值适用于"可选输入"场景。
defaults:: 软依赖的默认值映射 {dep_name: default_value}。软依赖未提供结果时使用。未在 defaults 中出现的软依赖默认为 None。

param 静态位置参数，追加在注入参数*之后*。:

kwargs:: 静态关键字参数。若与注入名冲突则抛出 InjectionError。
retry:: RetryPolicy 重试策略。默认仅尝试一次。
timeout:: 最大执行时长（秒）。None 表示不限制。异步任务使用 asyncio.wait_for()；同步任务通过线程 future 取消。
tags:: 自由标签，供 Graph.subgraph() 做选择性执行与调试，也可用于并发限制分组。
conditions:: 条件判断函数列表，接收依赖上下文，全部返回 True 时才执行任务。任一返回 False 则任务被标记为 SKIPPED。
cwd:: 工作目录。对 cmd 任务作为子进程工作目录；对 fn 任务通过临时切换当前目录生效。
env:: 环境变量覆盖映射。对 cmd 任务合并到子进程环境；对 fn 任务在执行期间临时设置。
verbose:: 是否打印详细输出。True 时打印执行的命令、返回码与输出（仅 cmd），以及任务生命周期。
skip_if_missing:: 仅对 cmd 为 list[str] 有效。True 时通过 shutil.which() 检查命令是否存在，不存在则跳过。
allow_upstream_skip:: 若为 True，硬依赖被 SKIPPED 时本任务仍执行（软依赖不影响）。适用于清理类任务。
strategy:: 单任务执行策略覆盖。None 表示继承图级策略。 "sequential" 同步直接调用；"thread"/"async" 将同步任务卸载到线程池，异步任务跑在事件循环上。
priority:: 同层任务调度优先级。数值越大越先启动。仅影响同层内启动顺序，不打破层屏障。默认 0。
concurrency_key:: 并发限制分组键。具有相同键的任务共享一个信号量，限制同时运行的实例数。具体限额由 run() 的 concurrency_limits 参数提供 {key: limit} 映射。None 表示不限制。
continue_on_error:: 若为 True，任务最终失败时不中止整图，仅标记本任务 FAILED，其硬依赖下游被 SKIPPED，其余任务继续。默认 False。
cache_key:: 缓存键计算函数。若提供，则用其基于依赖上下文计算的字符串键存取状态后端，使不同输入产生独立缓存条目。None 表示用任务名。
hooks:: TaskHooks 生命周期钩子。
executor:: 同步任务的执行器："thread"``（默认，线程池）/ ``"process" （进程池，绕过 GIL，适合 CPU 密集型；fn 须可 pickle）/ ``"inline"``（直接在事件循环线程调用，最快但会阻塞循环）。

name: str

fn: Callable[[...], T] | Callable[[...], Coroutine[Any, Any, T]] | None = None

cmd: List[str] | str | Callable[[...], Any] | None = None

depends_on: tuple[str, ...] = ()

soft_depends_on: tuple[str, ...] = ()

defaults: Mapping[str, Any]

retry: RetryPolicy

timeout: float | None = None

tags: tuple[str, ...] = ()

conditions: tuple[Callable[[Mapping[str, Any]], bool], ...] = ()

cwd: Path | None = None

env: Mapping[str, str] | None = None

verbose: bool = False

skip_if_missing: bool = False

allow_upstream_skip: bool = False

strategy: str | None = None

priority: int = 0

concurrency_key: str | None = None

continue_on_error: bool = False

cache_key: Callable[[Mapping[str, Any]], str] | None = None

hooks: TaskHooks

executor: str = 'thread'

property effective_fn: Callable[[...], T] | Callable[[...], Coroutine[Any, Any, T]]

获取有效的执行函数。

若提供 cmd，返回包装后的命令执行函数；否则返回 fn。包装函数在每次调用时从 self 读取 verbose/cwd/env/ timeout，避免闭包捕获运行期参数，使翻转字段无需重建 spec。

结果按实例缓存（functools.cached_property()）：frozen dataclass 字段不可变，_wrap_cmd 生成的闭包稳定，无需每次访问重建。

should_execute(context: Mapping[str, Any]) → tuple[bool, str | None][源代码]

检查任务是否应执行。

返回:: should_run 为 False 时 skip_reason 描述跳过原因。失败条件超过 2 个时仅展示前 2 个并附总数。
返回类型:: (should_run, skip_reason)

env_context() → ContextManager[None][源代码]

返回临时应用 env 与 cwd 的上下文管理器。

对 fn 任务生效。cmd 任务在 _run_command() 中直接传给子进程。

storage_key(context: Mapping[str, Any]) → str[源代码]: 计算状态后端存储键。

class pyflowx.RetryPolicy(max_attempts: int = 1, delay: float = 0.0, backoff: float = 1.0, jitter: float = 0.0, retry_on: tuple[type[BaseException], ...] = (<class 'Exception'>,))[源代码]

基类：object

任务失败重试策略。

参数

max_attempts:: 最大尝试次数（含首次）。1 表示仅尝试一次，不重试。
delay:: 两次尝试之间的初始等待秒数。
backoff:: 退避倍率。第 n 次重试等待 delay * backoff ** (n-1)。
jitter:: 抖动上限秒数。每次等待加上 [0, jitter) 的随机量，避免惊群。
retry_on:: 仅对这些异常类型重试。默认 (Exception,) 重试所有异常。传入空元组等价于不重试。

备注

替代旧版 retries: int。retries=2 等价于 RetryPolicy(max_attempts=3)。

max_attempts: int = 1

delay: float = 0.0

backoff: float = 1.0

jitter: float = 0.0

retry_on: tuple[type[BaseException], ...] = (<class 'Exception'>,)

property retries: int: 重试次数（不含首次），等价于 max_attempts - 1。

should_retry(exc: BaseException) → bool[源代码]: 异常是否属于可重试类型。

wait_seconds(attempt: int) → float[源代码]: 第 attempt 次失败后应等待的秒数（attempt 从 1 开始）。

class pyflowx.TaskHooks(pre_run: Callable[[TaskSpec[Any]], None] | None = None, post_run: Callable[[TaskSpec[Any], Any], None] | None = None, on_failure: Callable[[TaskSpec[Any], BaseException], None] | None = None)[源代码]

基类：object

任务生命周期钩子。

所有钩子均为可选。pre_run 在任务实际执行前调用；post_run 在成功后调用并接收返回值；on_failure 在最终失败后调用并接收异常。钩子异常不会影响任务状态，仅记录日志。

pre_run: Callable[[TaskSpec[Any]], None] | None = None

post_run: Callable[[TaskSpec[Any], Any], None] | None = None

on_failure: Callable[[TaskSpec[Any], BaseException], None] | None = None

class pyflowx.TaskStatus(value)[源代码]

基类：Enum

任务在单次运行内的生命周期状态。

PENDING = 'pending'

RUNNING = 'running'

SUCCESS = 'success'

FAILED = 'failed'

SKIPPED = 'skipped'

图构建

class pyflowx.Graph(specs: dict[str, ~pyflowx.task.TaskSpec[~typing.Any]]=<factory>, deps: dict[str, tuple[str, ...]]=<factory>, defaults: GraphDefaults = <factory>, namespace: str | None = None, _pending_refs: list[str] = <factory>, _resolved_cache: dict[str, ~pyflowx.task.TaskSpec[~typing.Any]]=<factory>)[源代码]

基类：object

校验后的有向无环任务图。

通过添加 TaskSpec 实例构建。每次 add 都执行即时校验（重名、缺失依赖），validate() / layers() 执行完整 DAG 校验（环检测）与拓扑分层。

图仅持有*配置*；运行时状态存于 RunReport。这使图可安全重复运行并在线程间共享。

specs: dict[str, TaskSpec[Any]]

deps: dict[str, tuple[str, ...]]

defaults: GraphDefaults

namespace: str | None = None

add(spec: TaskSpec[Any]) → Graph[源代码]: 注册一个任务 spec，并即时校验。返回 self 支持链式调用。

chain(*specs: TaskSpec[Any]) → Graph[源代码]

链式注册任务：每个 spec 自动依赖前一个。

chain(a, b, c) 等价于 b 依赖 a，c 依赖 b。若 spec 已带 depends_on，则前驱名追加到现有依赖前。返回 self 支持链式调用。

示例

>>> graph = px.Graph().chain(extract, transform, load)

add_subgraph(sub: Graph, *, namespace: str | None = None) → Graph[源代码]

将子图合并到当前图，任务名加命名空间前缀避免冲突。

参数

sub:: 待合并的子图。
namespace:: 命名空间前缀。None 时使用 sub.namespace，若子图也无命名空间则抛出 ValueError。最终任务名为 f"{ns}:{original_name}"。

合并后，子图内任务的依赖名也会被加前缀；与子图外部任务的依赖保持原样。

返回 self 支持链式调用。

validate() → None[源代码]: 执行完整 DAG 校验。存在环时抛出 CycleError。

property names: list[str]: 所有已注册任务名（按插入顺序）。

spec(name: str) → TaskSpec[Any][源代码]: 返回 name 的 spec；不存在则 KeyError。

resolved_spec(name: str) → TaskSpec[Any][源代码]

返回应用图级默认值后的 spec（不修改原图）。

对于 retry/timeout/strategy/env/cwd 等可空字段，若 spec 字段为默认空值且图级默认值非空，则用 dataclasses.replace() 生成带默认值的副本。

结果按 name 缓存；specs / defaults 变更时缓存失效。

dependencies(name: str) → tuple[str, ...][源代码]: name 的直接硬依赖前驱。

all_deps(name: str) → tuple[str, ...][源代码]: name 的硬依赖 + 软依赖。

all_specs() → Mapping[str, TaskSpec[Any]][源代码]: name -> spec 的只读视图。

layers() → list[list[str]][源代码]

将任务分组为可并行执行的层（Kahn 算法）。

同层任务无相互硬依赖，可并发执行。软依赖不参与分层。层按执行顺序返回。图有环时抛出 CycleError。

备注

本方法假定图已通过 validate() 校验（由 pyflowx.run() 在入口统一执行一次）。若直接调用本方法，需自行先校验。

subgraph(tags: Iterable[str]) → Graph[源代码]: 返回仅包含匹配任意标签的任务的新图。依赖边被修剪。

subgraph_by_names(names: Iterable[str]) → Graph[源代码]: 返回限定于 names 的新图（边已修剪）。

map(name_fn: Callable[[int], str], spec: TaskSpec[Any], items: Sequence[Any], arg_factory: Callable[[Any], tuple[Any, ...]] | None = None, depends_on_per: Callable[[int], tuple[str, ...]] | None = None) → list[TaskSpec[Any]][源代码]

为 items 中每个元素生成一个 TaskSpec 并加入图。

用于 fan-out / map-reduce 模式。返回生成的 spec 列表，便于后续 reduce 任务依赖。

参数:

name_fn -- 接受索引 i，返回任务名。需保证唯一。
spec -- 模板 spec。其 name 与 args 会被覆盖。
items -- 待分发的数据序列。
arg_factory -- 接受一个 item，返回位置参数元组，覆盖 spec.args。 None 则将单个 item 作为唯一位置参数。
depends_on_per -- 接受索引 i，返回该任务的额外硬依赖。None 则继承 spec.depends_on。

返回:

生成的 spec 列表（已加入图）。

返回类型:

list[TaskSpec]

示例

>>> fetch_tmpl = px.TaskSpec("", fn=fetch_user)
>>> specs = graph.map(lambda i: f"fetch_{i}", fetch_tmpl, [1, 2, 3])
>>> reduce_spec = px.TaskSpec("reduce", fn=reduce_fn, depends_on=tuple(s.name for s in specs))

to_mermaid(orientation: str = 'TD') → str[源代码]: 将 DAG 渲染为 Mermaid graph 定义字符串。

describe() → str[源代码]: 用于调试的人类可读多行摘要。

class pyflowx.GraphDefaults(retry: RetryPolicy | None = None, timeout: float | None = None, strategy: str | None = None, tags: tuple[str, ...] = (), env: Mapping[str, str] | None = None, cwd: Any = None, priority: int = 0, continue_on_error: bool = False, concurrency_key: str | None = None, verbose: bool = False)[源代码]

基类：object

图级默认值。TaskSpec 对应字段为 None 时回退到此处。

仅对可空字段生效（retry/timeout/strategy/env/cwd/tags/priority/ continue_on_error/concurrency_key）。非空字段（name/fn/cmd）不回退。

retry: RetryPolicy | None = None

timeout: float | None = None

strategy: str | None = None

tags: tuple[str, ...] = ()

env: Mapping[str, str] | None = None

cwd: Any = None

priority: int = 0

continue_on_error: bool = False

concurrency_key: str | None = None

verbose: bool = False

pyflowx.compose(graphs: dict[str, Graph]) → dict[str, Graph][源代码]

编程式解析多图的字符串引用，返回展开后的新图映射。

与 GraphComposer 等价，但作为独立函数暴露，供不使用 CliRunner 的编程式用户调用。

示例

>>> graphs = {
...     "build": px.Graph.from_specs([px.TaskSpec("b", cmd=["echo", "b"])]),
...     "all": px.Graph.from_specs(["build", px.TaskSpec("t", cmd=["echo", "t"])]),
... }
>>> resolved = px.compose(graphs)
>>> "b" in resolved["all"].all_specs()
True

创建任务模板工厂。

返回的工厂接受 name 与任意覆盖字段，生成 TaskSpec。适用于批量创建相似任务（如 fan-out）。

示例

>>> Fetch = px.task_template(fn=fetch_user, retry=px.RetryPolicy(max_attempts=3))
>>> specs = [Fetch(f"fetch_{uid}", args=(uid,)) for uid in range(5)]

执行

pyflowx.run(graph: Graph, strategy: Literal['sequential', 'thread', 'async', 'dependency'] = 'dependency', *, max_workers: int | None = None, dry_run: bool = False, verbose: bool = False, on_event: Callable[[TaskEvent], None] | None = None, state: StateBackend | None = None, concurrency_limits: Mapping[str, int] | None = None) → RunReport[源代码]

执行图并返回 RunReport。

参数

graph:: 待执行的已校验 Graph。
strategy:: 执行策略: "dependency"``（默认，依赖驱动无层屏障，最大并行度）/ ``"sequential" / "thread" / ``"async"``（层屏障模型）。
max_workers:: "thread" 的线程池大小。默认 min(32, len(layer))。
dry_run:: 若为 True，打印执行计划并返回空报告，不执行任务。
verbose:: 若为 True, 打印任务生命周期到 stdout。
on_event:: 可选回调，在每次状态转换时调用。
state:: 可选 StateBackend，用于断点续跑。
concurrency_limits:: {concurrency_key: max_concurrent} 映射。具有相同 concurrency_key 的任务共享信号量，限制同时运行实例数。

抛出

ValueError: strategy 不被识别时。
TaskFailedError: 任何任务耗尽重试后仍失败时（除非 continue_on_error=True）。

class pyflowx.RunReport(results: dict[str, ~pyflowx.task.TaskResult[~typing.Any]]=<factory>, success: bool = True)[源代码]

基类：object

工作流运行的聚合结果。

属性

results:: 任务名 -> TaskResult 的映射。插入顺序与任务完成顺序一致。
success:: 当且仅当所有非跳过任务都以 SUCCESS 结束时为 True。

results: dict[str, TaskResult[Any]]

success: bool = True

__getitem__(name: str) → Any[源代码]

返回任务 name 的*值*（而非 TaskResult）。

任务不在本次运行中则抛出 KeyError。未达到 SUCCESS 的任务返回 None。

result_of(name: str) → TaskResult[Any][源代码]: 返回 name 的完整 TaskResult。

summary() → dict[str, Any][源代码]: 用于日志/仪表盘的紧凑统计字典。

failed_tasks() → list[str][源代码]: 以 FAILED 状态结束的任务名列表。

succeeded_tasks() → list[str][源代码]: 以 SUCCESS 状态结束的任务名列表。

skipped_tasks() → list[str][源代码]: 以 SKIPPED 状态结束的任务名列表。

tasks_by_status(status: TaskStatus) → list[str][源代码]: 返回指定状态的任务名列表。

durations() → dict[str, float][源代码]: 任务名 -> 执行时长（秒）。无时长记录的为 0.0。

describe() → str[源代码]: 用于调试的人类可读多行报告。

class pyflowx.TaskResult(spec: TaskSpec[T], status: TaskStatus = TaskStatus.PENDING, value: T | None = None, error: BaseException | None = None, attempts: int = 0, started_at: datetime | None = None, finished_at: datetime | None = None, reason: str | None = None)[源代码]

基类：Generic[T]

运行期间产生的可变单任务记录。

spec: TaskSpec[T]

status: TaskStatus = 'pending'

value: T | None = None

error: BaseException | None = None

attempts: int = 0

started_at: datetime | None = None

finished_at: datetime | None = None

reason: str | None = None

property duration: float | None: 从开始到结束的耗时（秒），未开始/未结束则为 None。

YAML 编排

pyflowx.load_yaml(path: str | Path, variables: Mapping[str, Any] | None = None) → Graph[源代码]

从 YAML 文件加载任务图.

参数:

path (str | Path) -- YAML 文件路径
variables (Mapping[str, Any] | None) -- 运行时变量, 用于替换 ${VAR} 占位符

返回:

构建好的任务图

返回类型:

px.Graph

抛出:

YamlLoadError -- 文件不存在、YAML 格式错误、schema 校验失败、循环依赖等

pyflowx.parse_yaml_string(content: str, variables: Mapping[str, Any] | None = None) → Graph[源代码]

从 YAML 字符串构建任务图.

参数:

content (str) -- YAML 文本
variables (Mapping[str, Any] | None) -- 运行时变量, 用于替换 ${VAR} 占位符

返回:

构建好的任务图

返回类型:

px.Graph

加载 YAML 配置并执行任务图.

便捷函数, 组合 load_yaml() + (可选) 子图选择 + pyflowx.run(). 指定 jobs 时自动包含其传递依赖, 适用于从多 job 配置中执行单个命令.

参数:

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} 映射

返回:

执行报告

返回类型:

RunReport

抛出:

YamlLoadError -- YAML 加载或 schema 校验失败
KeyError -- jobs 指定了不存在的 job 名

pyflowx.run_cli(path: str | Path, argv: Sequence[str] | None = None) → int[源代码]

从 YAML 配置构建 CLI 并执行.

YamlCliRunner 的便捷包装.

参数:

path (str | Path) -- YAML 配置文件路径
argv (Sequence[str] | None) -- 命令行参数, None 表示使用 sys.argv[1:]

返回:

退出码 (0 成功, 1 失败)

返回类型:

int

pyflowx.build_cli_parser(data: Mapping[str, Any]) → Any[源代码]

从 YAML 配置的 cli: 段构建 argparse 解析器.

参数:: data (Mapping[str, Any]) -- 解析后的 YAML 配置 (包含 cli 和 jobs)
返回:: 构建好的参数解析器
返回类型:: argparse.ArgumentParser
抛出:: YamlLoadError -- cli 段格式错误

函数注册

pyflowx.register_fn(name: Callable[[P], T]) → Callable[[P], T][源代码]

pyflowx.register_fn(name: str | None = None) → Callable[[Callable[[P], T]], Callable[[P], T]]

装饰器：将函数注册到全局 registry.

支持两种用法:

@register_fn               # 使用函数 __name__ 作为注册名
def my_func(): ...

@register_fn("custom")     # 显式指定注册名
def my_func(): ...

参数:: name (str | Callable | None) -- 注册名或被装饰函数; 为 None 时使用函数 __name__
返回:: 装饰器函数或被装饰函数
返回类型:: Callable
抛出:: ValueError -- 名称已注册或无法推断函数名

pyflowx.get_fn(name: str) → Callable[[...], Any][源代码]

按名称获取已注册的函数.

参数:: name (str) -- 函数名
返回:: 已注册的函数
返回类型:: Callable
抛出:: KeyError -- 函数未注册

pyflowx.has_fn(name: str) → bool[源代码]

检查函数是否已注册.

参数:: name (str) -- 函数名
返回:: 是否已注册
返回类型:: bool

命令执行

pyflowx.run_command(spec: TaskSpec[Any]) → Any[源代码]

执行 spec.cmd 指定的命令（list / shell 字符串 / 可调用对象）。

与原 TaskSpec._run_command 行为一致：

可调用对象：直接调用，异常包装为 RuntimeError。
list / str：通过 subprocess.run() 执行，非零返回码抛 RuntimeError`（``verbose=False` 时附 stderr）。
verbose=True 时打印执行信息与返回码到 stdout。
cwd / env 通过 subprocess 参数隔离（进程级状态仅在 fn 任务路径使用，cmd 路径不依赖 os.chdir / os.environ）。

CLI 运行器

class pyflowx.CliRunner(aliases: dict[str, str | list[str | ~pyflowx.task.TaskSpec[~typing.Any]] | ~pyflowx.task.TaskSpec[~typing.Any] | ~pyflowx.graph.Graph]=<factory>, tasks: list[TaskSpec[Any]] = <factory>, strategy: Literal['sequential', 'thread', 'async', 'dependency']='dependency', description: str = <factory>, verbose: bool = <factory>)[源代码]

基类：object

命令行运行器: 根据用户输入执行对应的任务流图.

将命令别名映射到 Graph 实例. 通过 sys.argv 解析用户输入的命令, 执行对应的图.

参数:

aliases (dict[str, str | list[str] | Graph]) --
命令别名到任务引用的映射. 每个值可以是: * str —— 单个任务名 (引用 tasks 中注册的任务),

生成单任务图.
- list[str] —— 任务名列表, 自动 Graph.chain() 建立链式依赖, 即后一个任务依赖前一个.
- Graph —— 直接使用该图 (用于复杂场景, 如自定义 conditions、并行分支等).
tasks (list[TaskSpec]) -- 扁平注册的任务列表. aliases 中的字符串引用这些任务名. 未被任何 alias 引用的任务不会被执行.
strategy (str | Strategy) -- 默认执行策略. 可被命令行 --strategy 覆盖.
description (str) -- CLI 帮助文本.
verbose (bool) -- 是否显示详细执行过程. 默认 True, 可被命令行 --quiet 关闭.

示例

简单场景 (tasks + aliases):

runner = px.CliRunner(
    tasks=[
        px.cmd(["uv", "build"]),                      # name="uv_build"
        px.cmd(["maturin", "build"], name="maturin_build"),
        px.cmd(["ruff", "check", "--fix"], name="lint"),
    ],
    aliases={
        "b": "uv_build",
        "ba": ["uv_build", "maturin_build"],   # chain: maturin 依赖 uv
        "lint": "lint",
    },
)
runner.run()

复杂场景 (直接用 Graph):

runner = px.CliRunner(
    aliases={
        "a": px.Graph.from_specs([
            px.TaskSpec("add", cmd=["git", "add", "."], conditions=(...)),
            px.TaskSpec("commit", cmd=["git", "commit"], depends_on=("add",)),
        ]),
    },
)