Loading...
Loading...
Python design patterns for CLI scripts and utilities — type-first development, deep modules, complexity management, and red flags. Use when reading, writing, reviewing, or refactoring Python files, especially in .trellis/scripts/ or any CLI/scripting context. Also activate when planning module structure, deciding where to put new code, or doing code review.
npx skill4agent add mindfold-ai/trellis python-designDeep 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."""
...from dataclasses import dataclass
from typing import Literal
@dataclass(frozen=True)
class AgentRecord:
agent_id: str
task_name: str
worktree_path: Path
platform: Literal["claude", "codex", "cursor"]
status: Literal["running", "done", "failed"]
branch: strfrom 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)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():
passif data.get("status") == "running"# 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="claude")# 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.returncodedict# 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# 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)# 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 outputcommon/| Capability | Should Live In | Not In |
|---|---|---|
| JSON file read/write | | Each script's |
| Terminal colors + logging | | Each script's |
| Git command execution | | |
| Task data access | | Ad-hoc task.json parsing |
| Path constants | | Hardcoded strings |
_# 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 " "| Signal | What It Means |
|---|---|
| Shallow Module | Interface is nearly as complex as implementation |
| Information Leakage | Same JSON schema / file format knowledge in multiple modules |
| Duplicated Utility | Same helper function copied to multiple files |
| God Module | File > 500 lines with multiple unrelated responsibilities |
| Pass-Through Function | Function just forwards args to another with similar signature |
Magic | |
| sys.path Hacking | |
| Private-Named Public API | |
| Raw Dict Threading | Passing |
| Repeated Iteration | Same directory scan / file parse pattern in 3+ locations |
| Broad Exception Catch | |
| Temporal Decomposition | Modules split by "what runs when" instead of "what knows what" |
grep -r "pattern" .