如何使用Python-Click实现含源、任务、输出三阶段的多命令CLI？

完全可以实现你的需求。核心问题是默认的Click Group只允许执行单个子命令，而chain=True不支持嵌套Group作为子命令。我们可以通过自定义Group类来支持连续调用多个子命令，并强制验证三个阶段都被指定，同时保留每个阶段的可扩展性。

解决方案代码 #

import click

class SequentialGroup(click.Group):
    def parse_args(self, ctx, args):
        self.executed_commands = []
        while args:
            # 检查当前参数是否是顶层子命令（source/task/sink）
            if args[0] in self.commands:
                cmd_name = args.pop(0)
                cmd = self.commands[cmd_name]
                # 创建子上下文并解析该子命令的参数
                sub_ctx = ctx.copy()
                sub_ctx.parent = ctx
                cmd.parse_args(sub_ctx, args)
                # 执行子命令，将配置存入上下文
                cmd.invoke(sub_ctx)
                self.executed_commands.append(cmd_name)
            else:
                # 剩余参数不属于顶层子命令，终止解析
                break
        
        # 验证必须包含source、task、sink三个阶段
        required = {"source", "task", "sink"}
        missing = required - set(self.executed_commands)
        if missing:
            raise click.ClickException(f"缺少必填阶段: {', '.join(missing)}")

@click.group(cls=SequentialGroup)
@click.pass_context
def cli(ctx):
    # 初始化上下文存储，用于保存各阶段配置
    ctx.obj = {}

# -------------------------- Source 阶段 --------------------------
@cli.group()
@click.pass_context
def source(ctx):
    pass

@source.command()
@click.option("--read-size", default=4096, type=int, help="TCP读取缓冲区大小")
@click.argument("host")
@click.argument("port", type=int)
@click.pass_context
def tcp(ctx, read_size, host, port):
    """基于TCP套接字的输入源"""
    ctx.obj["source"] = {
        "type": "tcp",
        "read_size": read_size,
        "host": host,
        "port": port
    }

@source.command()
@click.argument("input_file", type=click.File("rb"))
@click.pass_context
def file(ctx, input_file):
    """基于文件的输入源"""
    ctx.obj["source"] = {
        "type": "file",
        "file": input_file
    }

# -------------------------- Task 阶段 --------------------------
@cli.group()
@click.pass_context
def task(ctx):
    pass

@task.command(name="taskABC")
@click.option("--optional-arg", default="default", help="可选参数")
@click.argument("arg1")
@click.argument("arg2", type=int)
@click.pass_context
def task_abc(ctx, optional_arg, arg1, arg2):
    """示例任务ABC，带可选参数和必填参数"""
    ctx.obj["task"] = {
        "type": "taskABC",
        "optional_arg": optional_arg,
        "arg1": arg1,
        "arg2": arg2
    }

@task.command()
@click.pass_context
def example1(ctx):
    """示例任务1"""
    ctx.obj["task"] = {"type": "example1"}

@task.command()
@click.pass_context
def example2(ctx):
    """示例任务2"""
    ctx.obj["task"] = {"type": "example2"}

# -------------------------- Sink 阶段 --------------------------
@cli.group()
@click.pass_context
def sink(ctx):
    pass

@sink.command()
@click.argument("host")
@click.argument("port", type=int)
@click.pass_context
def tcp(ctx, host, port):
    """基于TCP套接字的输出目标"""
    ctx.obj["sink"] = {
        "type": "tcp",
        "host": host,
        "port": port
    }

@sink.command()
@click.argument("output_file", type=click.File("wb"))
@click.pass_context
def file(ctx, output_file):
    """基于文件的输出目标"""
    ctx.obj["sink"] = {
        "type": "file",
        "file": output_file
    }

# -------------------------- 主执行逻辑 --------------------------
@cli.result_callback()
@click.pass_context
def run_pipeline(ctx, result, **kwargs):
    """收集各阶段配置，执行核心业务逻辑"""
    source_cfg = ctx.obj["source"]
    task_cfg = ctx.obj["task"]
    sink_cfg = ctx.obj["sink"]

    click.echo("=== 管道配置信息 ===")
    click.echo(f"输入源: {source_cfg}")
    click.echo(f"执行任务: {task_cfg}")
    click.echo(f"输出目标: {sink_cfg}")

    # 在这里添加你的实际业务逻辑：读取源数据 -> 执行任务 -> 写入输出
    # 例如：
    # data = read_source(source_cfg)
    # processed_data = run_task(task_cfg, data)
    # write_sink(sink_cfg, processed_data)

if __name__ == "__main__":
    cli()

关键特性说明 #

1. 自定义SequentialGroup

   • 重写parse_args方法，允许连续解析并执行多个顶层子命令（source/task/sink）
   • 自动验证三个阶段是否都被指定，缺少则抛出明确错误

2. 可扩展性

   • 每个阶段都是独立的Group，添加新类型只需新增对应Command：
     • 比如添加ZMQ输入源，只需在source Group下新增@source.command()修饰的函数
     • 新增任务类型同理，无需修改顶层核心逻辑

3. 上下文配置传递

   • 使用ctx.obj存储各阶段的配置信息，在最终的run_pipeline回调中统一获取并执行业务逻辑

测试你的示例命令 #

执行以下命令将正常运行并输出配置信息：

python myapp.py \
  source tcp --read-size=4096 127.0.0.1 8888 \
  task taskABC --optional-arg example example 99 \
  sink file myoutput.bin

如果缺少任意阶段（比如只指定source和task），会抛出错误：

Error: 缺少必填阶段: sink

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