subprocess Examples¶
[1]:
import os
import sys
import subprocess
dir_here = os.getcwd()
print(f"python interpreter: {sys.executable}")
print(f"python version: {sys.version_info}")
print(f"current dir: {dir_here}")
python interpreter: /Users/sanhehu/Documents/GitHub/Dev-Exp-Share/venv/bin/python
python version: sys.version_info(major=3, minor=8, micro=9, releaselevel='final', serial=0)
current dir: /Users/sanhehu/Documents/GitHub/Dev-Exp-Share/docs/source/02-SDE/01-Program-Language/02-Python-Root/Awesome-Python-Library-for-Software-Development/subprocess
[2]:
def print_result(res: subprocess.CompletedProcess):
print(f"type(res) = {type(res)!r}")
print(f"res = {res!r}")
print(f"res.args = {res.args!r}")
print(f"res.returncode = {res.returncode!r}")
print(f"res.stdout = {res.stdout!r}")
print(f"res.stderr = {res.stderr!r}")
Basic¶
Run Shell Command¶
[8]:
res = subprocess.run(["echo", "hello"])
print_result(res)
hello
type(res) = <class 'subprocess.CompletedProcess'>
res = CompletedProcess(args=['echo', 'hello'], returncode=0)
res.args = ['echo', 'hello']
res.returncode = 0
res.stdout = None
res.stderr = None
[9]:
res = subprocess.run(["python", "print_something.py"])
print_result(res)
hello alice
hello bob
type(res) = <class 'subprocess.CompletedProcess'>
res = CompletedProcess(args=['python', 'print_something.py'], returncode=0)
res.args = ['python', 'print_something.py']
res.returncode = 0
res.stdout = None
res.stderr = None
Handle Exit Code¶
出现错误时, 我们一般有两种处理方式:
无论命令是成功还是失败, 都继续运行. 不过我们希望能获得每个命令的状态码, 自己决定如何处理这些错误.
命令失败后立刻中断 Python 脚本, 阻止继续运行.
而 subprocess.run 的行为是这样的:
如果第一个命令比如
subprocess.run(["cat", ...])的这个 cat 本身就不存在, 找不到这个命令, 那么该行代码将不会被作为命令被执行, 而是直接丢出FileNotFoundErro从而停止后面的程序.而如果第一个命令存在, 那么如果是由于命令运行时导致的问题, 而你又没有加入
check=True参数, 则该行代码则会被执行然后继续执行后面的代码. 用户自己需要处理命令的状态码和错误.
[12]:
# command exit non-zero code, but subprocess.run just did and return result with error message
res = subprocess.run(["cat", "not-exists-file.txt"])
print_result(res) # this line still run
type(res) = <class 'subprocess.CompletedProcess'>
res = CompletedProcess(args=['cat', 'not-exists-file.txt'], returncode=1)
res.args = ['cat', 'not-exists-file.txt']
res.returncode = 1
res.stdout = None
res.stderr = None
cat: not-exists-file.txt: No such file or directory
[13]:
# raise exception immediately when command exit non-zero code
res = subprocess.run(["cat", "not-exists-file.txt"], check=True)
print("you should never see this line") # this line will not run
cat: not-exists-file.txt: No such file or directory
---------------------------------------------------------------------------
CalledProcessError Traceback (most recent call last)
Input In [13], in <cell line: 2>()
1 # raise exception immediately when command exit non-zero code
----> 2 res = subprocess.run(["cat", "not-exists-file.txt"], check=True)
3 print("you should never see this line")
File ~/.pyenv/versions/3.8.11/lib/python3.8/subprocess.py:516, in run(input, capture_output, timeout, check, *popenargs, **kwargs)
514 retcode = process.poll()
515 if check and retcode:
--> 516 raise CalledProcessError(retcode, process.args,
517 output=stdout, stderr=stderr)
518 return CompletedProcess(process.args, retcode, stdout, stderr)
CalledProcessError: Command '['cat', 'not-exists-file.txt']' returned non-zero exit status 1.
[4]:
res = subprocess.run(["ansible"])
print("you should never see this line") # this line will not run
---------------------------------------------------------------------------
FileNotFoundError Traceback (most recent call last)
Input In [4], in <cell line: 1>()
----> 1 res = subprocess.run(["ansible"])
2 print("you should never see this line")
File /Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/subprocess.py:493, in run(input, capture_output, timeout, check, *popenargs, **kwargs)
490 kwargs['stdout'] = PIPE
491 kwargs['stderr'] = PIPE
--> 493 with Popen(*popenargs, **kwargs) as process:
494 try:
495 stdout, stderr = process.communicate(input, timeout=timeout)
File /Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/subprocess.py:858, in Popen.__init__(self, args, bufsize, executable, stdin, stdout, stderr, preexec_fn, close_fds, shell, cwd, env, universal_newlines, startupinfo, creationflags, restore_signals, start_new_session, pass_fds, encoding, errors, text)
854 if self.text_mode:
855 self.stderr = io.TextIOWrapper(self.stderr,
856 encoding=encoding, errors=errors)
--> 858 self._execute_child(args, executable, preexec_fn, close_fds,
859 pass_fds, cwd, env,
860 startupinfo, creationflags, shell,
861 p2cread, p2cwrite,
862 c2pread, c2pwrite,
863 errread, errwrite,
864 restore_signals, start_new_session)
865 except:
866 # Cleanup if the child failed starting.
867 for f in filter(None, (self.stdin, self.stdout, self.stderr)):
File /Library/Developer/CommandLineTools/Library/Frameworks/Python3.framework/Versions/3.8/lib/python3.8/subprocess.py:1704, in Popen._execute_child(self, args, executable, preexec_fn, close_fds, pass_fds, cwd, env, startupinfo, creationflags, shell, p2cread, p2cwrite, c2pread, c2pwrite, errread, errwrite, restore_signals, start_new_session)
1702 if errno_num != 0:
1703 err_msg = os.strerror(errno_num)
-> 1704 raise child_exception_type(errno_num, err_msg, err_filename)
1705 raise child_exception_type(err_msg)
FileNotFoundError: [Errno 2] No such file or directory: 'ansible'
[ ]:
try:
res = subprocess.run(["ansible"])
except Exception as e:
print(e)
Capture Console Output¶
我们希望将 print 到 console output 的字符串捕获成一个变量, 然后对其进行处理.
[17]:
# capture console output (standard out)
res = subprocess.run(["echo", "hello"], capture_output=True)
res.stdout.decode("utf-8")
[17]:
'hello\n'
[18]:
res = subprocess.run(["python", "print_something.py"], capture_output=True)
res.stdout.decode("utf-8")
[18]:
'hello alice\nhello bob\n'
Pipe Pattern¶
Shell Command 里经常会出现 | pipe 语法, 把前一个命令的输出作为后一个命令的输入. 在 Python 中我们同样可以实现
[21]:
pipe = subprocess.Popen(["cat", "data.json"], stdout=subprocess.PIPE)
res = subprocess.run(["jq", ".version", "-r"], stdin=pipe.stdout, capture_output=True)
res.stdout.decode("utf-8").strip()
[21]:
'2.0'
Advance¶
run vs call vs Popen¶
subprocess.run: 3.5 中被加入, 是最高级的 API, 也是最推荐使用的 API, 返回的是结构化的 CompletedProcess 对象.
subprocess.call: 是 2.7 ~ 3.x+ 的 API, 是 Python2 时代的高级 API, 为了兼容性也一直存在着. 返回的是一个整数 return code.
subprocess.Popen: 是 2.7 ~ 3.x+ 的 API, 是 Python2 时代的高级 API, 为了兼容性也一直存在着. Popen 主要是创建一个管道, 然后 fork 一个子进程, 返回值在标准 IO 流中, 该管道用于父子进程通信. 父进程要么对子进程读, 要么写.
run() parameters¶
stdin:
stdout:
stderr:
input: 用于在父进程 pass 到 Popen.communicate() 方法, 然后变成子进程的 stdin
capture_output: bool, 是否捕获 console output
shell: bool / str, 如果为 True, 该命令会在一个 shell 中执行, 而 shell
cwd: 改变运行命令的目录
timeout: 只有在 Popen.communicate() 的时候才有用, 限制进程通信的时间上限
check: 如果 return code 不是 0, 则立刻抛出异常
encoding:
errors:
text:
env:
universal_newlines:
shell 参数详解
shell=True 意味着命令是在子进程中执行的, 你无法捕获到子进程的状态, 除非你显式的用 Popen 让子进程和父进程通信. 而且 shell=True 在当你允许用户输入自定义的 argument 的时候会有注注入风险, 导致潜在的安全问题.
该用法通常用于一个你的命令是做一些工作, 而你只想运行了就走, 不想管它运行的怎样, 结果如何的情况. 其他情况, 尽量避免用 shell=True.
另外该参数有时候会导致命令锁死, 例如 suprocess.run(["jq", "--version"], shell=True), 具体原因未知.
[27]:
res = subprocess.run(
[
"python", "print_something.py",
],
shell=True,
capture_output=True,
)
print_result(res)
type(res) = <class 'subprocess.CompletedProcess'>
res = CompletedProcess(args=['python', 'print_something.py'], returncode=0, stdout=b'', stderr=b'')
res.args = ['python', 'print_something.py']
res.returncode = 0
res.stdout = b''
res.stderr = b''
[29]:
# 你可以在 Python 中设定环境变量, 这个环境变量在 Python 运行时中都会生效
os.environ["my_var"] = "my_value"
res = subprocess.run(
[
"python", "print_my_var.py",
],
)
my_value
String Interpolation¶
在 shell 里你可以这么写, 用 $() 来做字符串替换: ````. 输出的结果
$ echo "Now is: $(date)"
Now is: Sun Jan 1 00:08:30 EDT 2022
但在 subprocess.run 里, 你应该直接用 Python 的字符串替换来完全替代 $(). 例子如下:
[13]:
_ = subprocess.run(
[
"echo",
"Now is: {}".format(
subprocess.run(["date"], capture_output=True).stdout.decode("utf-8")
)
]
)
Now is: Sun Jul 17 01:02:25 EDT 2022
[ ]: