TaskSpec —— 任务描述

TaskSpec 是不可变的任务描述符（Generic[T]，返回类型一路传到 RunReport），是唯一需要配置的东西。

主要参数说明：

px.TaskSpec(
    name="fetch_user",              # 唯一标识
    fn=fetch_user,                  # 同步或异步函数
    cmd=["curl", "..."],            # 或: 执行命令（覆盖 fn）
    depends_on=("auth",),           # 硬依赖（参与拓扑分层）
    soft_depends_on=("cache",),     # 软依赖（仅注入，不参与分层）
    args=(uid,),                    # 静态位置参数（追加在注入参数后）
    kwargs={"timeout": 30},         # 静态关键字参数
    retry=px.RetryPolicy(max_attempts=3, delay=1.0, backoff=2.0),
    timeout=30.0,                   # 超时秒数（None = 不限制）
    tags=("api", "user"),           # 自由标签，用于子图过滤
    conditions=(is_prod,),          # 条件函数列表（全部为 True 才执行）
    priority=10,                    # 同层内优先级（高优先执行，默认 0）
    concurrency_key="db",            # 并发分组键（配合 concurrency_limits 限流）
    cache_key=lambda ctx: str(ctx.get("uid")),  # 缓存键函数
    hooks=px.TaskHooks(pre_run=..., post_run=..., on_failure=...),
    cwd=Path("/tmp"),               # 命令工作目录（仅 cmd 模式）
    env={"DEBUG": "1"},             # 环境变量覆盖
    verbose=True,                   # 打印命令输出（仅 cmd 模式）
    skip_if_missing=True,           # 命令不存在时自动跳过（仅 list[str] cmd）
    allow_upstream_skip=False,      # 上游 SKIPPED/FAILED 时是否仍执行
    continue_on_error=False,        # 本任务失败是否不中断整体
)

两种任务形态

• **函数任务**（fn）：普通 Python 函数，参数名驱动自动注入

• **命令任务**（cmd）：执行外部命令，支持 list[str]、str``（shell）、``Callable 三种形态

skip_if_missing=True 时，list[str] 类型的 cmd 会通过 shutil.which 检查命令是否存在，不存在则跳过任务（标记为 SKIPPED）而非失败。

重试策略

RetryPolicy 配置重试次数、延迟、退避：

retry = px.RetryPolicy(
    max_attempts=3,    # 最大尝试次数
    delay=1.0,         # 初始延迟秒数
    backoff=2.0,       # 退避倍数
    jitter=0.1,        # 随机抖动（避免惊群）
    retry_on=(ConnectionError,),  # 仅对这些异常重试
)

任务钩子

TaskHooks 在任务生命周期触发（异常仅记录，不影响任务状态）：

hooks = px.TaskHooks(
    pre_run=lambda spec: print(f"start {spec.name}"),
    post_run=lambda spec, value: print(f"done {spec.name}"),
    on_failure=lambda spec, exc: alert(spec.name, exc),
)
px.TaskSpec("task", fn=work, hooks=hooks)

任务模板

task_template 工厂批量生成相似 TaskSpec：

fetch = px.task_template(
    fn=fetch_url,
    retry=px.RetryPolicy(max_attempts=5),
    timeout=30.0,
    tags=("api",),
)
graph = px.Graph.from_specs([
    fetch("users", url="https://api.example.com/users"),
    fetch("posts", url="https://api.example.com/posts"),
])

完整字段说明详见 API 参考。