python-design

CLI 脚本的 Python 设计

编写可维护的 Python CLI 工具和实用程序的设计模式与原则。 基于 A Philosophy of Software Design（Ousterhout），并针对脚本场景进行了调整。

何时启用

• 编写或修改 Python 文件
• 规划模块拆分
• 审查 Python 代码变更
• 重构感觉“杂乱”的脚本
• 添加新的子命令或实用函数

核心论点

核心挑战是管理复杂性，而不是添加功能。

复杂性是指任何让代码难以理解或修改的因素。它有三种表现：

1. 变更放大 — 一个小改动需要在多个位置编辑
2. 认知负荷 — 为了安全地进行修改，必须记住过多上下文
3. 未知的未知 — 你不知道自己不知道什么（最危险的一种）

复杂性是逐步累积的。它是数百个小决策积累的结果，而非某个灾难性错误。因此：小事也要认真对待。

---

原则 1：深模块

模块的价值取决于其隐藏功能与暴露接口的比例。

Deep module (good):          Shallow module (bad):
┌──────────┐                 ┌──────────────────────────┐
│ simple   │                 │ complex interface        │
│ interface│                 │ many params, many methods │
├──────────┤                 ├──────────────────────────┤
│          │                 │                          │
│  rich    │                 │  thin implementation     │
│  impl    │                 │                          │
│          │                 └──────────────────────────┘
│          │
└──────────┘

实用测试：如果调用者必须了解模块的内部工作方式才能正确使用它，那么该模块就太浅了。

示例：任务数据访问

# Shallow — caller must know JSON structure, file paths, error handling
def _read_json_file(path: Path) -> dict:
    with open(path, encoding="utf-8") as f:
        return json.load(f)

# Every caller does this independently:
task_path = tasks_dir / name / "task.json"
data = _read_json_file(task_path)
title = data.get("title") or data.get("name", "")
status = data.get("status", "planning")
assignee = data.get("assignee", "")

# Deep — caller gets what they need, module hides JSON/path/parsing
@dataclass(frozen=True)
class TaskInfo:
    name: str
    title: str
    status: str
    assignee: str
    priority: str
    directory: Path

def load_task(tasks_dir: Path, name: str) -> TaskInfo | None:
    """Load task by directory name. Returns None if not found."""
    ...

def list_active_tasks(tasks_dir: Path) -> list[TaskInfo]:
    """List all non-archived tasks, sorted by priority."""
    ...

深模块会吸收复杂性：JSON 解析、字段默认值、目录扫描、归档过滤。调用者只需处理类型化数据。

---

原则 2：类型优先开发

类型先于实现定义契约。以下工作流有助于及早发现设计问题：

1. 定义数据形状 — 先定义 dataclass 或 TypedDict
2. 定义函数签名 — 指定参数和返回值类型
3. 按类型要求实现 — 让类型检查器指导你补全实现
4. 在边界处验证 — 仅在数据进入系统时进行运行时检查

用于内部数据的冻结数据类

from dataclasses import dataclass
from typing import Literal

@dataclass(frozen=True)
class AgentRecord:
    agent_id: str
    task_name: str
    worktree_path: Path
    platform: Literal["Codex", "codex", "cursor"]
    status: Literal["running", "done", "failed"]
    branch: str

冻结数据类不可变，不会发生意外修改，传递起来也安全。

用于外部 JSON 形状的 TypedDict

如果数据来自文件（task.json、config.yaml、registry.json），请使用 TypedDict 记录预期的数据形状：

from typing import TypedDict, Required, NotRequired

class TaskData(TypedDict):
    title: Required[str]
    status: Required[str]
    assignee: NotRequired[str]
    priority: NotRequired[str]
    parent: NotRequired[str]
    children: NotRequired[list[str]]

这样就不必在各处散落 .get("field", default) 调用，数据形状只需记录一次。

用于领域基本类型的 NewType

当两个字符串代表不同的含义时，让类型系统强制区分它们：

from typing import NewType

TaskName = NewType("TaskName", str)    # directory name like "03-10-v040"
BranchName = NewType("BranchName", str)  # git branch like "feat/v0.4.0"

def create_branch(task: TaskName) -> BranchName:
    return BranchName(f"task/{task}")

用于状态的可辨识联合

当一个实体可能处于具有不同数据的多个状态时：

@dataclass(frozen=True)
class Pending:
    status: Literal["pending"] = "pending"

@dataclass(frozen=True)
class Running:
    status: Literal["running"] = "running"
    pid: int
    worktree: Path

@dataclass(frozen=True)
class Completed:
    status: Literal["completed"] = "completed"
    branch: str
    commit: str

AgentState = Pending | Running | Completed

def handle(state: AgentState) -> None:
    match state:
        case Running(pid=pid, worktree=wt):
            check_process(pid)
        case Completed(branch=br):
            create_pr(br)
        case Pending():
            pass

类型检查器会确保处理每种状态。不再需要编写 if data.get("status") == "running" 却遗漏某些分支。

---

原则 3：信息隐藏

每个模块都应封装设计决策。同一知识出现在多个模块中时，说明信息已经泄漏。

脚本中常见的信息泄漏模式

JSON 模式知识散落各处：

# BAD — 9 files all know how to iterate tasks and parse task.json
for d in sorted(tasks_dir.iterdir()):
    if d.name == "archive" or not d.is_dir():
        continue
    task_json = d / "task.json"
    if task_json.exists():
        data = json.loads(task_json.read_text())
        title = data.get("title") or data.get("name", "")
        ...

# GOOD — one module owns task iteration
# common/tasks.py
def iter_active_tasks(tasks_dir: Path) -> Iterator[TaskInfo]:
    """Yield all active (non-archived) tasks."""
    for d in sorted(tasks_dir.iterdir()):
        if d.name == "archive" or not d.is_dir():
            continue
        info = _load_task_json(d)
        if info:
            yield info

文件格式细节泄漏到各层：

# BAD — caller knows it's JSON, knows the path convention
registry_path = trellis_dir / "registry.json"
data = json.loads(registry_path.read_text())
data["agents"][agent_id] = {...}
registry_path.write_text(json.dumps(data, indent=2))

# GOOD — module hides storage format
registry = AgentRegistry(trellis_dir)
registry.add(agent_id, task=task_name, platform="Codex")

---

原则 4：将复杂性向下收拢

当复杂性无法避免时，模块应在内部吸收复杂性，而不是将其推给调用者。模块的开发者少、使用者多，因此由模块作者处理一次复杂性，胜过让每个调用者各自处理。

# BAD — pushes complexity to every caller
def run_git(args: list[str]) -> subprocess.CompletedProcess:
    return subprocess.run(["git"] + args, capture_output=True, text=True)

# Every caller must: check returncode, decode stderr, handle encoding,
# strip whitespace, handle repo not found, etc.

# GOOD — absorbs complexity
def run_git(args: list[str], *, cwd: Path | None = None) -> str:
    """Run git command, return stdout. Raises GitError on failure."""
    result = subprocess.run(
        ["git"] + args,
        capture_output=True, text=True, encoding="utf-8",
        errors="replace", cwd=cwd,
    )
    if result.returncode != 0:
        raise GitError(args[0], result.stderr.strip())
    return result.stdout.strip()

将复杂性上推的反模式

• 返回原始的 subprocess.CompletedProcess，让调用者检查 .returncode
• 引发通用异常，让调用者自行解析
• 使用配置参数来回避决策
• 在类型化对象能让调用者省去验证的情况下返回 dict

---

原则 5：让错误不复存在

异常处理是复杂性的主要来源。最佳策略是通过设计语义，让错误条件不再是错误。

# BAD — raises if key doesn't exist
def remove_agent(registry: dict, agent_id: str) -> None:
    if agent_id not in registry["agents"]:
        raise KeyError(f"Agent {agent_id} not found")
    del registry["agents"][agent_id]

# GOOD — guarantees postcondition: agent is not in registry
def remove_agent(registry: dict, agent_id: str) -> None:
    """Ensure agent_id is not in the registry after this call."""
    registry["agents"].pop(agent_id, None)

# BAD — raises if directory already exists
def init_workspace(path: Path) -> None:
    if path.exists():
        raise FileExistsError(f"{path} already exists")
    path.mkdir()

# GOOD — guarantees postcondition: directory exists
def ensure_workspace(path: Path) -> Path:
    """Ensure workspace directory exists. Returns the path."""
    path.mkdir(parents=True, exist_ok=True)
    return path

关键在于：根据操作的后置条件（“调用后，X 为真”）定义操作，而不是根据其前置条件（“调用前，X 必须为真”）定义。

---

原则 6：KISS 和三次法则

KISS — 保持简单

选择能满足需求的最简单方案。复杂性必须由具体（而非假设的）需求来证明其合理性。

# Over-engineered — registry pattern for 3 formatters
class FormatterRegistry:
    _registry: dict[str, type] = {}
    @classmethod
    def register(cls, name: str): ...
    @classmethod
    def create(cls, name: str): ...

# Simple — just a dictionary
FORMATTERS = {"json": format_json, "text": format_text, "table": format_table}

def format_output(fmt: str, data: Any) -> str:
    formatter = FORMATTERS.get(fmt)
    if not formatter:
        raise ValueError(f"Unknown format: {fmt}")
    return formatter(data)

三次原则

在一种模式出现 三次 之前，先不要提取抽象。两次只是巧合，三次才是模式。过早抽象比重复更糟糕，因为它会：

• 通过共享抽象将无关代码耦合起来
• 让每个实例都更难独立理解
• 使人倾向于把未来的情况硬塞进抽象，即使它们并不适用

不过：一旦确实出现三次，就立即提取。不要等到九次。

---

原则 7：单一职责与模块边界

每个模块都应该只有 一个变更原因。当模块超过约 300 行时，请检查它是否承担了多项职责。

拆分信号

出现以下情况时应进行拆分：

• 文件中有多个由注释标题分隔的“部分”
• 你只需要从一个大型模块中导入一个函数
• 模块不同部分的测试没有共用的准备工作
• 修改一项职责时无需了解另一项职责

如何拆分

按 信息隐藏（封装了哪些知识）拆分，而非按执行顺序（何时运行）拆分。

# BAD — split by execution order (temporal decomposition)
# step1_parse_args.py, step2_validate.py, step3_execute.py
# All three must know the command structure

# GOOD — split by responsibility
# task_store.py    — owns task.json read/write, schema, iteration
# task_cli.py      — owns argparse, subcommand routing
# task_display.py  — owns formatting, colors, table output

---

原则 8：一致的共享基础设施

多个脚本需要相同的功能时，应在 common/ 中集中提供一次。

功能	应放在	不应放在
JSON 文件读写	common/io.py	每个脚本各自的 _read_json_file
终端颜色 + 日志记录	common/log.py	每个脚本各自的 Colors 类
Git 命令执行	common/git.py	以 _run_git_command 为前缀的私有函数
任务数据访问	common/tasks.py	临时解析 task.json
路径常量	common/paths.py（现有文件）	硬编码字符串

命名：如果其他模块会使用某个函数，它就是公共 API——不要在名称前加 _。

---

原则 9：结构化 CLI 输出解析

解析 shell 命令（git、grep 等）的输出时，应遵循语义空白：

# BAD — .strip() destroys semantic whitespace
# git submodule status prefix: ' ' = initialized, '-' = uninitialized, '+' = changed
line = output_line.strip()  # Loses the prefix character!

# GOOD — strip only trailing newlines
line = output_line.rstrip("\n\r")
prefix = line[0] if line else " "

解析结构化命令输出时，始终说明每个字段位置的含义。

---

危险信号速查

代码审查和自我审查时使用：

信号	含义
浅层模块	接口几乎和实现一样复杂
信息泄漏	多个模块中都了解相同的 JSON 模式或文件格式
重复工具函数	同一个辅助函数被复制到多个文件
上帝模块	文件超过 500 行，且承担多项无关职责
透传函数	函数只是将参数转发给签名相似的另一个函数
魔法 .get() 链	data.get("x") or data.get("y", "") — 缺少类型定义
篡改 sys.path	sys.path.insert(0, ...) — 应改为修复包结构
以私有命名的公共 API	_function 被 3 个或更多外部模块导入
原始字典传递	在 4 个或更多函数调用中传递 dict — 应使用 dataclass
重复迭代	相同的目录扫描或文件解析模式出现在 3 个或更多位置
宽泛捕获异常	使用 except Exception: 却不重新抛出 — 会掩盖错误
按时间顺序拆分	模块按“何时运行”而非“了解什么”来拆分

---

设计检查清单（编写代码前）

1. 先定义类型：编写逻辑之前，先定义数据形态
2. 检查模块深度：接口是否比实现更简单？
3. 扫描重复项：创建新工具函数前，先运行 grep -r "pattern" .
4. 检查职责：这是否应归属某个现有模块？
5. 错误设计：能否通过设计消除这种错误？
6. 精准命名：无需阅读实现，名称是否就能传达含义？

设计检查清单（代码审查期间）

1. 扫描危险信号：对照差异检查上表
2. 类型安全：是否为新的数据结构记录了类型？
3. 信息隐藏：此次更改是否泄露了实现细节？
4. 一致性：是否遵循了模块中现有的模式？
5. 深度：调用方使用常见路径是否简单？

---

战略性投入

每次更改都投入约 10-20% 的精力来改善周边设计。

代码能够运行是必要条件，但并非充分条件。软件开发的增量应该是抽象，而不仅仅是功能。每次更改都应该让代码库比之前好一点。

这不是完美主义，而是复利。细小的设计改进会逐渐累积，让系统随着时间推移变得极易使用。