Typer 异步命令与选项

示例说明

Typer 把类型注解变成命令行接口：参数类型决定解析与帮助文档，Enum 直接变成 choices 列表，envvar 让敏感值从环境注入而非命令行历史留痕。异步命令天然支持 await——批处理里逐批推进并输出进度，asyncio.run 包装整个应用入口即可。

技术要点

• 类型注解驱动选项定义与帮助生成
• envvar 为敏感值提供环境通道
• 范围校验在解析期即拒绝非法值

代码解读

replicate 命令的地区参数是枚举限定三种取值；批次大小带区间校验并可从环境变量取默认；api 隐藏输入且强制存在；dry-run 与 execute 互斥开关用组合语法声明。命令内逐批 sleep 模拟异步推进。status 演示枚举默认值。

使用建议

敏感选项一律隐藏输入并优先环境变量。异步命令里别做 CPU 密集工作会卡事件循环。帮助文本认真写，它是 CLI 的第一文档。

代码示例

import asyncio
from enum import Enum

import typer

app = typer.Typer(add_completion=False)


class Region(str, Enum):
    eu = 'eu'
    na = 'na'
    apac = 'apac'


@app.command()
async def replicate(
    region: Region = typer.Option(..., help='target region'),
    batch_size: int = typer.Option(500, min=1, max=5000, envvar='REPL_BATCH_SIZE'),
    api_key: str = typer.Option(..., envvar='REPLICATE_API_KEY', hide_input=True),
    dry_run: bool = typer.Option(False, '--dry-run/--execute'),
) -> None:
    """Replicate catalog entries into a regional store."""
    typer.echo(f'replicating to {region.value} with batch={batch_size}')
    if dry_run:
        typer.echo('dry run: no rows will be written')
        return

    for batch_index in range(3):
        await asyncio.sleep(0.1)
        typer.echo(f'batch {batch_index + 1}/3 done')

    typer.echo('replication complete')


@app.command()
def status(region: Region = typer.Option(Region.eu)) -> None:
    typer.echo(f'worker status in {region.value}: idle')


def main() -> None:
    asyncio.run(app())


if __name__ == '__main__':
    main()