wehub-skill-sync
67 lines
2.0 KiB
Markdown
67 lines
2.0 KiB
Markdown
---
|
|
|
|
# 可运行示例：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")` 注入仅当应用程序在同一进程中运行时才有效。如果使用子进程，则可能需要通过环境变量或配置文件传递测试数据。
|