subprocess 模块

短命女 2021-12-21 20:51 266阅读 0赞

*  subprocess 模块学习
    
     *  subprocess.run()
     *  subprocess.CompletedProcess
     *  subprocess.DEVNULL
     *  subprocess.PIPE
     *  subprocess.STDOUT
     *  Popen constructor

## subprocess 模块学习 ##

> [链接][Link 1]

推荐通过run()来创建进程，更为高级的，可以使用Popen.

### subprocess.run() ###

> subprocess.run(args, \*, stdin=None, input=None, stdout=None, stderr=None, capture\_output=False, shell=False, cwd=None, timeout=None, check=False, encoding=None, errors=None, text=None, env=None, universal\_newlines=None)

下面介绍的参数，大部分也适用于其他关于subprocess的情况.  
args表示我们需要运行的一些参数和子程序的名字等等, 而stdin, stdout, stderr分别表示标准输入、标准输出和标准错误.

#### capture\_output ####

如果capture\_output=True，那么标准输出stdout和标准错误stderr会被捕获:

""" subcontrol.py print("???") raise TypeError("error here") """ def main(): child = os.path.join(os.path.dirname(__file__), "subcontrol.py") pipes = [] s = "See you again, Robot {0}" command = [sys.executable, child] p1 = subprocess.run(command, capture_output=True) p2 = subprocess.run(command, stdin=subprocess.PIPE, stderr=subprocess.STDOUT) print(p1) print(p2) if __name__ == "__main__": main()

输出为:

???
    Traceback (most recent call last):
      File "C:/Py\subcontrol.py", line 10, in <module>
        raise TypeError("error here")
    TypeError: error here
    CompletedProcess(args=['C:\\Ana\\python.exe', 'C:/Py\\subcontrol.py'], returncode=1, stdout=b'???\r\n', stderr=b'Traceback (most recent call last):\r\n  File "C:/Py\\subcontrol.py", line 10, in <module>\r\n    raise TypeError("error here")\r\nTypeError: error here\r\n')
    CompletedProcess(args=['C:\\Ana\\python.exe', 'C:/Py\\subcontrol.py'], returncode=1)

上面的结果总，前俩个元素都是由p2带来的，我们可以看到，标准输出和标准错误确实被捕获了.  
capture\_output=True其实等价于设置为stdout=subprocess.PIPE, stderr=subprocess.PIPE, 所以，这几个参数并不能共存.

p2 = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

#### timeout ####

如果timeout不是时间，而是像下面的一样给了时间(所以单位是秒？), 如果子程序运行的时间超过了限定，那么就会报出TimeoutExpired错误, 而且这个参数是通过传递给Popen.communicate()来实现的.

""" subcontrol.py import time print("begin the subprocess") time.sleep(3) print("end the subprocess") raise TypeError("error here") """ def main(): child = os.path.join(os.path.dirname(__file__), "subcontrol.py") pipes = [] s = "See you again, Robot {0}" command = [sys.executable, child] p1 = subprocess.run(command, timeout=2) print(p1) if __name__ == "__main__": main()

结果是

begin the subprocess
    Traceback (most recent call last):
      File "C:\Ana\lib\subprocess.py", line 468, in run
        stdout, stderr = process.communicate(input, timeout=timeout)
      File "C:\Ana\lib\subprocess.py", line 952, in communicate
        sts = self.wait(timeout=self._remaining_time(endtime))
      File "C:\Ana\lib\subprocess.py", line 984, in wait
        return self._wait(timeout=timeout)
      File "C:\Ana\lib\subprocess.py", line 1226, in _wait
        raise TimeoutExpired(self.args, timeout)
    subprocess.TimeoutExpired: Command '['C:\\Ana\\python.exe', 'C:/Py\\subcontrol.py']' timed out after 2.0 seconds
    
    During handling of the above exception, another exception occurred:
    
    Traceback (most recent call last):
      File "C:/Py/maincontrol.py", line 21, in <module>
        main()
      File "C:/Py/maincontrol.py", line 15, in main
        p1 = subprocess.run(command, timeout=2)
      File "C:\Ana\lib\subprocess.py", line 473, in run
        stderr=stderr)
    subprocess.TimeoutExpired: Command '['C:\\Ana\\python.exe', 'C:/Py\\subcontrol.py']' timed out after 2 seconds

#### input ####

input也是通过传递给Popen.communcate()来实现的，而且表示stdin=subprocess.PIPE.

""" import time import sys print("begin the subprocess") stdin = sys.stdin.buffer.read() lines = stdin.decode("utf8", "ignore") time.sleep(1) print(lines) print("end the subprocess") """ def main(): child = os.path.join(os.path.dirname(__file__), "subcontrol.py") pipes = [] s = "See you again, Robot {0}" command = [sys.executable, child] p1 = subprocess.run(command, input=b"eric") print(p1) if __name__ == "__main__": main()

begin the subprocess
    eric
    end the subprocess
    CompletedProcess(args=['C:\\Ana\\python.exe', 'C:/Py\\subcontrol.py'], returncode=0)

#### check ####

如果check为True且returncode非零(0表示子程序运行成功)， 那么一个CalledProcessError错误会被报出.

""" import time import sys print("begin the subprocess") stdin = sys.stdin.buffer.read() lines = stdin.decode("utf8", "ignore") time.sleep(1) print(lines) print("end the subprocess") raise TypeError("error here") """ def main(): child = os.path.join(os.path.dirname(__file__), "subcontrol.py") pipes = [] s = "See you again, Robot {0}" command = [sys.executable, child] p1 = subprocess.run(command, input=b"eric", check=True) print(p1) if __name__ == "__main__": main()

begin the subprocess
    eric
    end the subprocess
    Traceback (most recent call last):
      File "C:/Py\subcontrol.py", line 17, in <module>
        raise TypeError("error here")
    TypeError: error here
    Traceback (most recent call last):
      File "C:/Py/maincontrol.py", line 21, in <module>
        main()
      File "C:/Py/maincontrol.py", line 15, in main
        p1 = subprocess.run(command, input=b"eric", check=True)
      File "C:\Ana\lib\subprocess.py", line 481, in run
        output=stdout, stderr=stderr)
    subprocess.CalledProcessError: Command '['C:\\Ana\\python.exe', 'C:/Py\\subcontrol.py']' returned non-zero exit status 1.

#### encoding, error, text ####

我们可以指定encoding, error 或者text, 使得打开文件的时候，以特定的方式打开（以及错误？）.

import subprocess import os, sys def main(): child = os.path.join(os.path.dirname(__file__), "subcontrol.py") pipes = [] s = "See you again, Robot {0}" command = [sys.executable, child] p1 = subprocess.run(command, input="eric", encoding="utf8") #or text=True print(p1) if __name__ == "__main__": main()

begin the subprocess
    Traceback (most recent call last):
    eric
      File "C:/Py\subcontrol.py", line 17, in <module>
    end the subprocess
        raise TypeError("error here")
    TypeError: error here
    CompletedProcess(args=['C:\\Ana\\python.exe', 'C:/Py\\subcontrol.py'], returncode=1)

#### env ####

env参数应当为一个映射，如果为None, 则子程序会继承原环境变量.  
不知道怎么弄额.

### subprocess.CompletedProcess ###

subprocess.run()会返回completedprocess对象.

### subprocess.DEVNULL ###

这个不晓得啥意思, 应该也是指定一种输入输出的方式吧.

### subprocess.PIPE ###

例如stdout=subprocess.PIPE, 感觉就是把输出捕获了，所以就是定义了一种输入输出的方式.

### subprocess.STDOUT ###

专门针对stderr，好像即使使得stderr和stdout有同样的输出方式:

第一种情况，我们不加任何修饰:

def main(): child = os.path.join(os.path.dirname(__file__), "subcontrol.py") command = [sys.executable, child] p1 = subprocess.run(command, input=b"eric") print(p1) if __name__ == "__main__": main()

![在这里插入图片描述][20190626220004732.png]

第二种，我们令stderr=subprocess.STDOUT

![在这里插入图片描述][20190626220057704.png]  
此时，报错部分应该也是被视作输出的一部分所以并没有显示红色.

再加一个stdout=subprocess.PIPE:

CompletedProcess(args=['C:\\Ana\\python.exe', 'C:/Py\\subcontrol.py'], returncode=1, stdout=b'begin the subprocess\r\neric\r\nTraceback (most recent call last):\r\n  File "C:/Py\\subcontrol.py", line 15, in <module>\r\n    raise TypeError("ddd")\r\nTypeError: ddd\r\n')

只返回了一个completeprocess对象，而且这个对象只包含stdout而没有stderr，如果令stderr=subprocess.PIPE, 那么completeprocess对象是会包含stderr的.

### Popen constructor ###

> class subprocess.Popen(args, bufsize=-1, executable=None, stdin=None, stdout=None, stderr=None, preexec\_fn=None, close\_fds=True, shell=False, cwd=None, env=None, universal\_newlines=None, startupinfo=None, creationflags=0, restore\_signals=True, start\_new\_session=False, pass\_fds=(), \*, encoding=None, errors=None, text=None)

我们可以通过subprocess.Popen来创建进程，里面多了许多不同的参数，我也没仔细去看，重点还是介绍一下其方法.

#### Popen.poll() ####

检查子进程是否结束如果是就返回returncode，否则返回None

def main(): child = os.path.join(os.path.dirname(__file__), "subcontrol.py") command = [sys.executable, child] p1 = subprocess.Popen(command) print(p1.poll()) print(p1) if __name__ == "__main__": main()

None
    <subprocess.Popen object at 0x000001C56874CCF8>

如果子程序结束了，返回的是0.

#### Popen.wait(timeout=None) ####

等待程序至结束并返回returncode，如果超时，报TimeoutExpired错误.

def main(): child = os.path.join(os.path.dirname(__file__), "subcontrol.py") command = [sys.executable, child] p1 = subprocess.Popen(command) print(p1.wait()) print(p1.poll()) print(p1) if __name__ == "__main__": main()

begin the subprocess
    end the subprocess
    0
    0
    <subprocess.Popen object at 0x000001DA1C80CD30>

#### Popen.communicate(input=None, timeout=None) ####

这个方法用于与子程序进行交互，我们可以通过其向子程序传入数据，并获取其stdout和stderr.

communicate()会返回(stdout\_data, stderr\_data)供我们使用，需要注意的是，输入和所获得输出是受模式和编码限制的，默认的都是二进制.

另外，我们需要使得stdin=subprocess.PIPE, 如果想要获得stdout，则stdout=subprocess.PIPE, stderr也是一样的.

def main(): child = os.path.join(os.path.dirname(__file__), "subcontrol.py") command = [sys.executable, child] p1 = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE) out = p1.communicate(input=b"eric") print(p1.poll()) print(out) print(p1) if __name__ == "__main__": main()

0
    (b'begin the subprocess\r\neric\r\nend the subprocess\r\n', None)
    <subprocess.Popen object at 0x0000023379917828>

注意如果没有stdin=subprocess.PIPE, 程序好像会挂死，如果没有stdout=subprocess.PIPE, 结果会是(None, None).

#### Popen.send\_signal(signal) ####

信号是啥啊？

#### Popen.terminate() ####

结束子程序.

#### Popen.pid ####

返回子程序的id:

def main(): child = os.path.join(os.path.dirname(__file__), "subcontrol.py") command = [sys.executable, child] p1 = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE) p1.communicate(b"eric") print(p1.pid) print(p1.poll()) print(p1) if __name__ == "__main__": main()

当然popen.stdin, popen.stdout, popen.stderr, popen.returncode这些属性也都是存在的.

剩下的就不看了.

转载于:https://www.cnblogs.com/MTandHJ/p/11094506.html

[Link 1]: https://docs.python.org/3/library/subprocess.html
[20190626220004732.png]: /images/20211220/73f5eaaf6293452e81c62017ee874ff0.png
[20190626220057704.png]: /images/20211220/1db311c8fd65403798d14454a9f2f62a.png