Click 链式子命令编排

示例说明

数据管道 CLI 的自然形态是 extract、transform、load 串联执行——链式命令组允许一条命令行写多个子命令，按输入顺序依次执行。共享状态放进上下文对象，各步骤读写同一家账本；result_callback 在全部步骤结束后统一收尾输出，避免每个子命令各自为政。执行顺序约束显式校验，乱序即拒绝。

技术要点

• chain=True 启用子命令串联执行
• ctx.obj 作为跨步骤共享账本
• result_callback 统一收尾输出

代码解读

cli 组开启链模式并把 verbose 与结果列表放进 obj；extract 校验源路径存在后记录步骤；transform 先检查前置步骤已执行，再按标志转换数据；result_callback 遍历结果列表输出整条流水线的执行摘要。

使用建议

复杂管道考虑显式 DAG 描述而非命令顺序。共享状态要防步骤间隐式耦合，键名约定清晰。退出码按步骤成败设置供 CI 判断。

代码示例

import click


@click.group(chain=True)
@click.option('--verbose', is_flag=True, help='enable verbose output')
@click.pass_context
def cli(ctx, verbose):
    ctx.ensure_object(dict)
    ctx.obj['verbose'] = verbose
    ctx.obj['results'] = []


@cli.command('extract')
@click.argument('source', type=click.Path(exists=True))
@click.pass_obj
def extract(state, source):
    click.echo(f'extracting {source}')
    state['results'].append(('extract', source))


@cli.command('transform')
@click.option('--uppercase', is_flag=True)
@click.pass_obj
def transform(state, uppercase):
    if not any(step == 'extract' for step, _ in state['results']):
        raise click.UsageError('transform must run after extract')
    value = 'transformed-data'
    if uppercase:
        value = value.upper()
    state['results'].append(('transform', value))
    click.echo(f'transform ok: {value}')


@cli.result_callback()
def process_pipeline(pipeline, state, **kwargs):
    click.echo('pipeline finished:')
    for step, value in state['results']:
        click.echo(f'  {step} -> {value}')


if __name__ == '__main__':
    cli()