subprocess 模塊

subprocess 模塊學習

鏈接

推薦通過run()來創(chuàng)建進程，更為高級的，可以使用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)

下面介紹的參數(shù)，大部分也適用于其他關于subprocess的情況.
args表示我們需要運行的一些參數(shù)和子程序的名字等等, 而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, 所以，這幾個參數(shù)并不能共存.

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

timeout

如果timeout不是時間，而是像下面的一樣給了時間(所以單位是秒？), 如果子程序運行的時間超過了限定，那么就會報出TimeoutExpired錯誤, 而且這個參數(shù)是通過傳遞給Popen.communicate()來實現(xiàn)的.

"""
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()來實現(xiàn)的，而且表示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參數(shù)應當為一個映射，如果為None, 則子程序會繼承原環(huán)境變量.
不知道怎么弄額.

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來創(chuàng)建進程，里面多了許多不同的參數(shù)，我也沒仔細去看，重點還是介紹一下其方法.

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)

這個方法用于與子程序進行交互，我們可以通過其向子程序傳入數(shù)據(jù)，并獲取其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)

信號是啥?。?/p>

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這些屬性也都是存在的.

剩下的就不看了.

?著作權歸作者所有,轉載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務。

相關閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容