526互联

Python执行命令的正确做法

发布时间 2023-09-07 02:50:04作者: ffl

在编写Python程序的时候，很容易直接调用system, subprocess.Popen, subprocess.run, subprocess.call, subprocess.call_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)

用法：

  1. 用来使用 system 的地方换成 Executor.system, 原来使用 subprocess.xxx 的地方换成 Executor.xxx，参数完全一样。
  2. 可以通过 Executor.init 和 Executor.finish 来开始和结束，结束的时候把命令序列都记录到一个日志文件里，用以分析。
  3. 还可以通过调用 Executor.run_phony("@some split") 方法来插入一些分隔行，便于问题诊断。phony 是Makefile里 .PHONY 一样的意思，就是“伪造的，假的”的意思。