系統管理腳本常需執行 Shell 命令。本文設計 Shell 類別封裝 subprocess.Popen,支援同步/非同步執行、狀態查詢、超時終止、遠端 SSH 執行、切換使用者 (su/sudo) 等進階功能,並提供完整測試用例。
Shell
class Shell:
def __init__(self, cmd):
self.cmd = cmd
self.ret_code = self.ret_info = self.err_info = None
self._process = None
def run_background(self):
self._process = subprocess.Popen(self.cmd, shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
def run(self):
self.run_background()
self.wait()
def wait(self):
self.ret_info, self.err_info = self._process.communicate()
self.ret_code = self._process.returncode
def get_status(self):
return "RUNNING" if self._process.poll() is None else "FINISHED"
def kill(self):
os.kill(self._process.pid, signal.SIGKILL)
self.wait() # 必須等待回收
RemoteShell(cmd, ip):透過 SSH 執行遠端命令,自動處理金鑰認證與逾時。SuShell(cmd, user):以 su - user -c "cmd" 切換使用者 (需 root)。SuShellDeprecated:使用 fork+setuid (不推薦,環境變數不乾淨)。
# 同步執行
sh = Shell('ls -l /tmp')
sh.run()
print(sh.ret_code, sh.ret_info)
# 非同步 + 進度輪詢
sh = Shell('sleep 10')
sh.run_background()
while sh.get_status() == "RUNNING":
time.sleep(1)
print("running...")
sh.wait()
# 遠端執行
rs = RemoteShell('hostname', '192.168.1.10')
rs.run()
# 切換使用者
ss = SuShell('whoami', 'www-data')
ss.run()
stdout/stderr 使用管道,大量輸出可能阻塞,生產環境建議改用執行緒讀取或臨時檔案。kill() 後必須呼叫 wait() 回收殭屍進程。StrictHostKeyChecking=no 避免首次連線提示,ConnectTimeout 防止卡死。提供 10+ 測試案例:正常命令、錯誤命令、非同步、終止、多命令、遠端連線、IP 錯誤、無信任關係、切換使用者成功/失敗/需密碼等。
此封裝類別可直接整合至自動化部署、監控、備份等運維腳本。