寻求轻量Python库实现链式任务执行与结果传递

绝对有！针对你需要的链式任务执行+结果传递场景（pre-run验证 → 执行 → post-run处理），我推荐几个比Celery、Luigi轻量得多的开箱即用库，完全适配你的需求：

1. Prefect Open Source #

Prefect的核心版本非常轻量，不需要搭建复杂的集群或服务，单进程就能运行。它天然支持任务依赖定义、自动结果传递，还能处理失败重试、状态追踪这些实用功能，完全覆盖你的三个任务阶段。

举个简单的示例：

from prefect import task, Flow

@task
def pre_run_validation(input_data):
    # 这里写你的验证逻辑
    if input_data is None:
        raise ValueError("输入数据不能为空")
    return {"validated": True, "data": input_data}

@task
def execute(validation_result):
    # 接收验证结果，执行核心逻辑
    processed_data = validation_result["data"] * 2  # 示例处理
    return {"processed": processed_data}

@task
def post_run(execution_result):
    # 接收执行结果，做后续处理（比如记录、清理）
    print(f"任务完成，处理结果：{execution_result['processed']}")
    return "post-run处理完成"

# 定义任务流，设置链式依赖
with Flow("任务执行流") as flow:
    validation_out = pre_run_validation({"key": "value"})
    execution_out = execute(validation_out)
    post_run(execution_out)

# 运行流
flow.run()

2. Invoke #

Invoke是一个极简的任务执行库，主打轻量和易用性，没有多余的依赖。它虽然主要用于命令行任务，但可以非常灵活地实现链式调用，手动传递任务结果，完全满足你的需求——毕竟你的场景就是线性的三个步骤，用Invoke写起来特别直观。

示例代码：

from invoke import task

@task
def pre_run_validation(c, input_data):
    # c是Invoke的上下文对象，不需要命令行参数时可忽略
    if not input_data:
        raise Exception("验证失败：输入为空")
    return {"status": "valid", "payload": input_data}

@task
def execute(c, validation_result):
    processed = validation_result["payload"] + "_processed"
    return {"result": processed}

@task
def post_run(c, execution_result):
    print(f"Post-run处理：记录结果 {execution_result['result']}")
    return "done"

# 链式调用，手动传递结果
if __name__ == "__main__":
    validation_out = pre_run_validation(None, {"input": "test"})
    execution_out = execute(None, validation_out)
    post_run(None, execution_out)

3. PyFlow #

PyFlow是一个专门针对轻量级任务流的库，支持可视化任务依赖（可选），并且天然支持任务间的结果传递。它的语法非常直观，适合快速搭建简单的链式任务流。

示例代码：

from pyflow import Workflow

def pre_run_validation(input_data):
    assert input_data is not None, "验证失败"
    return input_data

def execute(data):
    return data * 3

def post_run(result):
    print(f"最终结果：{result}")

# 创建工作流，定义任务顺序
wf = Workflow()
pre_task = wf.add(pre_run_validation, args=("初始数据",))
exec_task = wf.add(execute, args=(pre_task.output,))
post_task = wf.add(post_run, args=(exec_task.output,))

# 运行工作流
wf.run()

如果你需要的是超轻量，甚至不想引入额外库，其实用普通的Python函数链式调用也能搞定——不过既然你问的是开箱即用的库，上面这三个都是非常合适的选择，比Celery、Luigi轻便太多，不需要复杂的配置就能跑起来。

内容的提问来源于stack exchange，提问作者user14809632