3.3 subprocess - 執行外部命令
3.3 subprocess - 執行外部命令
subprocess 模組讓你從 Python 程式中執行外部命令。在 Hook 系統中,主要用於執行 Git 命令。
基本用法
subprocess.run()
最推薦的方式,Python 3.5+ 引入:
1import subprocess
2
3# 基本執行
4result = subprocess.run(["ls", "-la"])
5
6# 捕獲輸出
7result = subprocess.run(
8 ["ls", "-la"],
9 capture_output=True,
10 text=True
11)
12print(result.stdout) # 標準輸出
13print(result.stderr) # 錯誤輸出
14print(result.returncode) # 返回碼實際範例:Git 操作
來自 .claude/lib/git_utils.py:
1import subprocess
2from typing import Optional
3
4def run_git_command(
5 args: list[str],
6 cwd: Optional[str] = None,
7 timeout: int = 10
8) -> tuple[bool, str]:
9 """
10 執行 git 命令並返回結果
11
12 Args:
13 args: git 命令參數列表(不含 'git')
14 cwd: 執行目錄,預設為當前目錄
15 timeout: 命令超時時間(秒)
16
17 Returns:
18 tuple[bool, str]: (是否成功, 輸出內容或錯誤訊息)
19 """
20 try:
21 result = subprocess.run(
22 ["git"] + args,
23 cwd=cwd,
24 capture_output=True,
25 text=True,
26 timeout=timeout
27 )
28 if result.returncode == 0:
29 return True, result.stdout.strip()
30 else:
31 return False, result.stderr.strip()
32 except subprocess.TimeoutExpired:
33 return False, f"Command timed out after {timeout}s"
34 except FileNotFoundError:
35 return False, "git command not found"
36 except Exception as e:
37 return False, str(e)使用範例:
1# 取得當前分支
2success, output = run_git_command(["branch", "--show-current"])
3if success:
4 print(f"Current branch: {output}")
5
6# 取得專案根目錄
7success, output = run_git_command(["rev-parse", "--show-toplevel"])
8
9# 取得 worktree 列表
10success, output = run_git_command(["worktree", "list", "--porcelain"])重要參數
capture_output
捕獲標準輸出和錯誤輸出:
1# 捕獲輸出
2result = subprocess.run(
3 ["git", "status"],
4 capture_output=True, # 等同於 stdout=PIPE, stderr=PIPE
5 text=True
6)
7
8# 等價寫法
9result = subprocess.run(
10 ["git", "status"],
11 stdout=subprocess.PIPE,
12 stderr=subprocess.PIPE,
13 text=True
14)text
將輸出解碼為字串(而非 bytes):
1# text=False(預設)
2result = subprocess.run(["echo", "hello"], capture_output=True)
3print(result.stdout) # b'hello\n'(bytes)
4
5# text=True
6result = subprocess.run(["echo", "hello"], capture_output=True, text=True)
7print(result.stdout) # 'hello\n'(str)cwd
指定工作目錄:
1result = subprocess.run(
2 ["git", "status"],
3 cwd="/path/to/repo",
4 capture_output=True,
5 text=True
6)timeout
設定超時時間(秒):
1try:
2 result = subprocess.run(
3 ["long_running_command"],
4 timeout=10,
5 capture_output=True
6 )
7except subprocess.TimeoutExpired:
8 print("Command timed out!")check
自動檢查返回碼:
1try:
2 # 如果返回碼非零,拋出 CalledProcessError
3 result = subprocess.run(
4 ["git", "status"],
5 check=True,
6 capture_output=True,
7 text=True
8 )
9except subprocess.CalledProcessError as e:
10 print(f"Command failed with return code {e.returncode}")
11 print(f"Error output: {e.stderr}")錯誤處理
常見異常
1import subprocess
2
3def safe_run_command(args: list[str]) -> tuple[bool, str]:
4 try:
5 result = subprocess.run(
6 args,
7 capture_output=True,
8 text=True,
9 timeout=30
10 )
11 if result.returncode == 0:
12 return True, result.stdout.strip()
13 return False, result.stderr.strip()
14
15 except subprocess.TimeoutExpired:
16 return False, "Command timed out"
17
18 except FileNotFoundError:
19 return False, f"Command not found: {args[0]}"
20
21 except PermissionError:
22 return False, f"Permission denied: {args[0]}"
23
24 except Exception as e:
25 return False, str(e)安全考量
避免 shell=True
1# 危險:使用者輸入可能導致命令注入
2user_input = "file.txt; rm -rf /"
3subprocess.run(f"cat {user_input}", shell=True) # 危險!
4
5# 安全:使用列表傳遞參數
6subprocess.run(["cat", user_input]) # 安全驗證輸入
1def run_git_command(args: list[str]) -> tuple[bool, str]:
2 # 驗證第一個參數是否為有效的 git 子命令
3 valid_commands = ["status", "branch", "log", "diff", "rev-parse"]
4 if args and args[0] not in valid_commands:
5 return False, f"Invalid git command: {args[0]}"
6 # ...進階用法
管道(Pipe)
1# 模擬 "ls -la | grep .py"
2ls_process = subprocess.run(
3 ["ls", "-la"],
4 capture_output=True,
5 text=True
6)
7
8grep_process = subprocess.run(
9 ["grep", ".py"],
10 input=ls_process.stdout,
11 capture_output=True,
12 text=True
13)環境變數
1import os
2
3# 自訂環境變數
4env = os.environ.copy()
5env["MY_VAR"] = "value"
6
7result = subprocess.run(
8 ["my_script.sh"],
9 env=env,
10 capture_output=True
11)實際應用:分支資訊
1def get_current_branch() -> Optional[str]:
2 """獲取當前分支名稱"""
3 success, output = run_git_command(["branch", "--show-current"])
4 return output if success and output else None
5
6def get_project_root() -> str:
7 """獲取專案根目錄"""
8 success, output = run_git_command(["rev-parse", "--show-toplevel"])
9 return output if success else os.getcwd()
10
11def get_worktree_list() -> list[dict]:
12 """獲取所有 worktree 列表"""
13 success, output = run_git_command(["worktree", "list", "--porcelain"])
14 if not success:
15 return []
16
17 worktrees = []
18 current_worktree = {}
19
20 for line in output.split("\n"):
21 if line.startswith("worktree "):
22 if current_worktree:
23 worktrees.append(current_worktree)
24 current_worktree = {"path": line[9:]} # 移除 "worktree " 前綴
25 elif line.startswith("branch "):
26 branch_ref = line[7:] # 移除 "branch " 前綴
27 if branch_ref.startswith("refs/heads/"):
28 branch_ref = branch_ref[11:]
29 current_worktree["branch"] = branch_ref
30
31 if current_worktree:
32 worktrees.append(current_worktree)
33
34 return worktrees最佳實踐
1. 使用列表而非字串
1# 好
2subprocess.run(["git", "commit", "-m", "Fix bug"])
3
4# 不好
5subprocess.run("git commit -m 'Fix bug'", shell=True)2. 設定合理的超時
1# 為長時間運行的命令設定超時
2subprocess.run(["long_task"], timeout=300) # 5 分鐘3. 處理所有可能的錯誤
1def robust_run(args):
2 try:
3 result = subprocess.run(args, ...)
4 return True, result.stdout
5 except subprocess.TimeoutExpired:
6 return False, "Timeout"
7 except FileNotFoundError:
8 return False, "Command not found"
9 except Exception as e:
10 return False, str(e)思考題
- 為什麼
run_git_command返回tuple[bool, str]而不是直接拋出異常? shell=True有什麼風險?什麼情況下必須使用?- 如何實作一個可以同時執行多個命令的函式?
實作練習
- 寫一個函式,執行命令並即時顯示輸出(不等待完成)
- 實作一個命令重試機制(失敗時自動重試 N 次)
- 寫一個函式,執行多個命令並收集所有結果
上一章:json - 序列化 下一章:re - 正規表達式