Python执行命令的正确做法
在编写 Python 程序的时候,很容易直接调用 system, subprocess.Popen, subprocess.run, subprocess.call, subprocess.check_call, subprocess.check_output 等方法执行命令。但是如果一个系统里充满了这样的命令之后,整个系统变得难以分析和调试,在编程里就是所谓的「可观察性」很差。或者说,这样的脚本系统不是「Hackable」的。
有什么解法呢?有的,根据我多年手写此类脚本系统的经验,就是要把原子的命令执行统一记录和 dump 下来,这样在系统执行完的时候,只要分析这样的命令序列文件就能快速定位问题。是不是和汇编很像?其实也和数据库的SQL 语句日志文件很像。
下面的代码就可以做到:
import os
import subprocess
import json
from loguru import logger
g_all_commands = []
class Executor:
@staticmethod
def init():
g_all_commands = []
@staticmethod
def finish(log_file):
logger.info(f'\n')
logger.info(f'<commands>: --------------------')
logger.info(f'<commands>: all commands')
logger.info(f'<commands>: --------------------')
logger.info(f'<commands>: {log_file}')
logger.info(f'<commands>: --------------------')
logger.info(f'\n')
with open(log_file,'w') as f:
f.writelines([c+'\n' for c in g_all_commands])
@staticmethod
def run_phony(cmd):
g_all_commands.append(cmd)
@staticmethod
def system(command):
g_all_commands.append(command)
return os.system(command)
@staticmethod
def run(*popenargs, input=None, capture_output=False, timeout=None, check=False, **kwargs):
g_all_commands.append(Executor.__join(popenargs))
return subprocess.run(*popenargs, input=input, capture_output=capture_output, timeout=timeout, check=check, **kwargs)
@staticmethod
def Popen(*args, **kwargs):
g_all_commands.append(Executor.__join(args))
return subprocess.Popen(*args, **kwargs)
@staticmethod
def check_output(*popenargs, timeout=None, **kwargs):
g_all_commands.append(Executor.__join(popenargs))
return subprocess.check_output(*popenargs, timeout, **kwargs)
@staticmethod
def call(*popenargs, timeout=None, **kwargs):
g_all_commands.append(Executor.__join(popenargs))
return subprocess.call(*popenargs, timeout, **kwargs)
@staticmethod
def check_call(*popenargs, **kwargs):
g_all_commands.append(Executor.__join(popenargs))
return subprocess.check_call(*popenargs, **kwargs)
@staticmethod
def __join(*popenargs):
a = []
for e in popenargs:
if type(e)==type({}):
a.append(json.dumps(e))
elif type(e)==type([]) or type(e)==type(()):
a.append(Executor.__join(*e))
else:
a.append(str(e))
return ' '.join(a)
用法:
- 用来使用
system的地方换成Executor.system, 原来使用subprocess.xxx的地方换成Executor.xxx,参数完全一样。 - 可以通过
Executor.init和Executor.finish来开始和结束,结束的时候把命令序列都记录到一个日志文件里,用以分析。 - 还可以通过调用
Executor.run_phony("@some split")方法来插入一些分隔行,便于问题诊断。phony是 Makefile 里.PHONY一样的意思,就是“伪造的,假的”的意思。