--- # 可运行示例:CLI 应用程序 **当应用程序从命令行被调用时**(例如 `python -m myapp`,使用 argparse/click 的 CLI 工具)。 **做法**:使用 `asyncio.create_subprocess_exec` 调用 CLI 并捕获输出。 ```python # pixie_qa/run_app.py import asyncio import sys from pydantic import BaseModel import pixie class AppArgs(BaseModel): query: str class AppRunnable(pixie.Runnable[AppArgs]): """通过子进程驱动 CLI 应用程序。""" @classmethod def create(cls) -> "AppRunnable": return cls() async def run(self, args: AppArgs) -> None: proc = await asyncio.create_subprocess_exec( sys.executable, "-m", "myapp", "--query", args.query, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=120) if proc.returncode != 0: raise RuntimeError(f"App 失败(退出码 {proc.returncode}):{stderr.decode()}") ``` ## 当 CLI 需要修补的依赖项时 如果 CLI 读取外部服务,请创建一个包装入口点,在运行真实 CLI 之前修补依赖项: ```python # pixie_qa/patched_app.py """在运行真实 CLI 之前修补外部依赖项的入口点。""" import myapp.config as config config.redis_url = "mock://localhost" from myapp.main import main main() ``` 然后将 Runnable 指向该包装器: ```python async def run(self, args: AppArgs) -> None: proc = await asyncio.create_subprocess_exec( sys.executable, "-m", "pixie_qa.patched_app", "--query", args.query, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=120) ``` **注意**:对于 CLI 应用程序,`wrap(purpose="input")` 注入仅当应用程序在同一进程中运行时才有效。如果使用子进程,则可能需要通过环境变量或配置文件传递测试数据。