subprocess 模块

Posted 2022-02-14 mtandhj

目录

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

subprocess 模块学习

链接

推荐通过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()

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

此时，报错部分应该也是被视作输出的一部分所以并没有显示红色.

再加一个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这些属性也都是存在的.

剩下的就不看了.