提问人:freshWoWer 提问时间:9/18/2008 最后编辑:John SmithfreshWoWer 更新时间:7/24/2023 访问量:4532313
如何执行程序或调用系统命令?
How do I execute a program or call a system command?
答:
import os
os.system("your command")
请注意,这很危险,因为该命令未被清除。请参阅 os
和 sys
模块的文档。有一堆函数(exec
* 和 spawn*
)可以做类似的事情。
评论
subprocess
import os
cmd = 'ls -al'
os.system(cmd)
如果要返回命令的结果,可以使用 os.popen
。但是,自 2.6 版以来,这已被弃用,取而代之的是 subprocess 模块,其他答案已经很好地涵盖了该模块。
评论
使用 subprocess.run
:
import subprocess
subprocess.run(["ls", "-l"])
另一种常见的方法是 os.system
,但你不应该使用它,因为如果命令的任何部分来自程序外部或可能包含空格或其他特殊字符,它是不安全的,通常也更灵活(你可以获得 stdout
、stderr
、“真实”状态代码、更好的错误处理等)。甚至 os.system
的文档也建议改用。subprocess.run
subprocess
在 Python 3.4 及更早版本上,使用 代替 :subprocess.call
.run
subprocess.call(["ls", "-l"])
评论
echo $PATH
call(["echo", "$PATH"])
$PATH
shell=True
shell=True
os.path.expandvars("$PATH")
pip list | grep anatome
subprocess.run(["balcon.exe","-n","Tatyana","-t", "Hello world"])
我建议使用 subprocess 模块而不是 os.system,因为它会为您进行 shell 转义,因此更安全。
subprocess.call(['ping', 'localhost'])
评论
subprocess
shell=False
shlex.split
)
使用子进程。
...或者对于一个非常简单的命令:
import os
os.system('cat testfile')
os.system
还可以,但有点过时了。它也不是很安全。相反,请尝试 . 不直接调用 sh,因此比 更安全。subprocess
subprocess
os.system
在此处获取更多信息。
评论
subprocess
以下是调用外部程序的方法的摘要,包括它们的优缺点:
os.system
将命令和参数传递到系统的 shell。这很好,因为您实际上可以以这种方式一次运行多个命令,并设置管道和输入/输出重定向。例如:os.system("some_command < input_file | another_command > output_file")
但是,虽然这很方便,但您必须手动处理 shell 字符(如空格等)的转义。另一方面,这也允许您运行简单的 shell 命令,而不是实际的外部程序。
os.popen
将执行相同的操作,只是它为您提供了一个类似文件的对象,您可以使用该对象访问该过程的标准输入/输出。popen 还有其他 3 种变体,它们处理 I/O 的方式略有不同。如果将所有内容作为字符串传递,则命令将传递到 shell;如果你把它们作为一个列表传递,那么你就不需要担心转义任何东西。例:os.system
print(os.popen("ls -l").read())
子进程。噗噗
。这是 的替代品,但缺点是由于过于全面而稍微复杂一些。例如,你会说:os.popen
print subprocess.Popen("echo Hello World", shell=True, stdout=subprocess.PIPE).stdout.read()
而不是
print os.popen("echo Hello World").read()
但是,将所有选项都放在一个统一的类中,而不是 4 个不同的 popen 函数中,这很好。请参阅文档。
subprocess.call
。 这基本上就像类一样,接受所有相同的参数,但它只是等到命令完成并给你返回码。例如:Popen
return_code = subprocess.call("echo Hello World", shell=True)
subprocess.run
. 仅限 Python 3.5+。与上述类似,但更加灵活,并在命令完成执行时返回CompletedProcess
对象。os.fork
、 与它们的 C 语言对应物类似,但我不建议直接使用它们。os.exec
os.spawn
该模块可能应该是您使用的模块。subprocess
最后,请注意,对于所有方法,您将 shell 要执行的最终命令作为字符串传递,并且您负责对其进行转义。如果传递的字符串的任何部分不能完全信任,则存在严重的安全隐患。例如,如果用户正在输入字符串的某些/任何部分。如果不确定,请仅将这些方法与常量一起使用。为了给您一个含义的提示,请考虑以下代码:
print subprocess.Popen("echo %s " % user_input, stdout=PIPE).stdout.read()
想象一下,用户输入了一些可以擦除整个文件系统的“”内容。my mama didnt love me && rm -rf /
评论
open
subprocess.run()
subprocess.run(..)
subprocess.check_output(..)
for
subprocess.run()
subprocess.check_call()
subprocess.Popen()
os.popen()
典型实现:
import subprocess
p = subprocess.Popen('ls', shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
for line in p.stdout.readlines():
print line,
retval = p.wait()
您可以自由地对管道中的数据执行您想做的事情。事实上,您可以简单地省略这些参数(和),它的行为将类似于 。stdout
stdout=
stderr=
os.system()
评论
.readlines()
一次读取所有行,即它阻塞,直到子进程退出(关闭管道的末端)。要实时读取(如果没有缓冲问题),您可以:for line in iter(p.stdout.readline, ''): print line,
p.stdout.readline()
s
Popen
shell=True
subprocess.run()
这里还有另一个区别,前面没有提到。
subprocess.Popen
将 <命令>作为子进程执行。就我而言,我需要执行文件<a>它需要与另一个程序<b>通信。
我尝试了子进程,执行成功。但是,<b>无法与<a>通信。 当我从终端运行两者时,一切都很正常。
还有一个: (注意:kwrite 的行为与其他应用程序不同。如果您在Firefox上尝试以下操作,结果将不相同。
如果尝试,程序流将冻结,直到用户关闭 kwrite。为了克服这个问题,我尝试了.这一次程序继续流动,但 kwrite 成为控制台的子进程。os.system("kwrite")
os.system(konsole -e kwrite)
任何人运行的 kwrite 都不是子进程(即在系统监视器中,它必须出现在树的最左边)。
评论
subprocess
有关将子进程与调用进程分离的一些提示(在后台启动子进程)。
假设您想从 CGI 脚本开始一个长任务。也就是说,子进程的生存时间应比 CGI 脚本执行进程长。
子流程模块文档中的经典示例如下:
import subprocess
import sys
# Some code here
pid = subprocess.Popen([sys.executable, "longtask.py"]) # Call subprocess
# Some more code here
这里的想法是,您不想在“调用子进程”行中等待,直到 longtask.py 完成。但目前尚不清楚示例中的“这里还有更多代码”行之后会发生什么。
我的目标平台是 FreeBSD,但开发是在 Windows 上,所以我首先在 Windows 上遇到了这个问题。
在 Windows (Windows XP) 上,在 longtask.py 完成其工作之前,父进程不会完成。这不是您想要的 CGI 脚本。该问题并非特定于 Python;在PHP社区中,问题是一样的。
解决方案是将进程创建标志传递给 Windows API 中的基础 CreateProcess 函数DETACHED_PROCESS。 如果你碰巧安装了pywin32,你可以从win32process模块导入标志,否则你应该自己定义它:
DETACHED_PROCESS = 0x00000008
pid = subprocess.Popen([sys.executable, "longtask.py"],
creationflags=DETACHED_PROCESS).pid
/* UPD 2015.10.27 @eryksun在下面的评论中指出,语义上正确的标志是 CREATE_NEW_CONSOLE (0x00000010) */
在 FreeBSD 上,我们还有另一个问题:当父进程完成时,它也会完成子进程。这也不是你想要的 CGI 脚本。一些实验表明,问题似乎出在共享 sys.stdout 上。有效的解决方案如下:
pid = subprocess.Popen([sys.executable, "longtask.py"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE)
我没有检查过其他平台上的代码,也不知道 FreeBSD 上这种行为的原因。如果有人知道,请分享您的想法。在 Python 中搜索启动后台进程还没有提供任何信息。
评论
import subprocess as sp;sp.Popen('calc')
cmd
calc
DETACHED_PROCESS
creationflags
CREATE_NEW_CONSOLE
os.devnull
还要检查“pexpect”Python 库。
它允许对外部程序/命令进行交互式控制,甚至 ssh、ftp、telnet 等。你可以输入如下内容:
child = pexpect.spawn('ftp 192.168.0.24')
child.expect('(?i)name .*: ')
child.sendline('anonymous')
child.expect('(?i)password')
subprocess.check_call
如果您不想测试返回值,则很方便。它会在任何错误时引发异常。
如果您需要正在调用的命令的输出, 那么你可以使用 subprocess.check_output (Python 2.7+)。
>>> subprocess.check_output(["ls", "-l", "/dev/null"])
'crw-rw-rw- 1 root root 1, 3 Oct 18 2007 /dev/null\n'
另请注意 shell 参数。
如果 shell 是 ,则指定的命令将通过 shell 执行。如果您使用 Python 主要是为了增强它对大多数系统 shell 提供的控制流,并且仍然希望方便地访问其他 shell 功能,例如 shell 管道、文件名通配符、环境变量扩展以及将 ~ 扩展到用户的主目录,这将非常有用。但是,请注意,Python 本身提供了许多类似 shell 的功能的实现(特别是 、 、 、 、 和 )。
True
glob
fnmatch
os.walk()
os.path.expandvars()
os.path.expanduser()
shutil
评论
check_output
subprocess.check_output("ls -l /dev/null".split())
shell=True
.split()
shlex.split()
我总是用来做这些事情。下面是一个演示代码:fabric
from fabric.operations import local
result = local('ls', capture=True)
print "Content:/n%s" % (result, )
但这似乎是一个很好的工具:sh
(Python 子进程接口)。
请看一个例子:
from sh import vgdisplay
print vgdisplay()
print vgdisplay('-v')
print vgdisplay(v=True)
os.system
不允许您存储结果,因此,如果您想将结果存储在某个列表或其他东西中,则可以使用。subprocess.call
您可以使用 Popen,然后可以检查过程的状态:
from subprocess import Popen
proc = Popen(['ls', '-l'])
if proc.poll() is None:
proc.kill()
查看子流程。噗噗。
运行任何命令并返回结果的最简单方法:
from commands import getstatusoutput
try:
return getstatusoutput("ls -ltr")
except Exception, e:
return None
评论
的命令
文档说它在 2.6 中被弃用,并将在 3.0 中删除。
这就是我运行命令的方式。这段代码几乎包含了您需要的一切
from subprocess import Popen, PIPE
cmd = "ls -l ~/"
p = Popen(cmd , shell=True, stdout=PIPE, stderr=PIPE)
out, err = p.communicate()
print "Return code: ", p.returncode
print out.rstrip(), err.rstrip()
评论
更新:
subprocess.run
从 Python 3.5 开始,如果您的代码不需要保持与早期 Python 版本的兼容性,则建议使用这种方法。它更一致,并提供与 Envoy 相似的易用性。(不过,管道并不那么简单。请参阅此问题了解如何操作。
下面是文档中的一些示例。
运行进程:
>>> subprocess.run(["ls", "-l"]) # Doesn't capture output
CompletedProcess(args=['ls', '-l'], returncode=0)
在失败的运行中引发:
>>> subprocess.run("exit 1", shell=True, check=True)
Traceback (most recent call last):
...
subprocess.CalledProcessError: Command 'exit 1' returned non-zero exit status 1
捕获输出:
>>> subprocess.run(["ls", "-l", "/dev/null"], stdout=subprocess.PIPE)
CompletedProcess(args=['ls', '-l', '/dev/null'], returncode=0,
stdout=b'crw-rw-rw- 1 root root 1, 3 Jan 23 16:23 /dev/null\n')
原答案:
我建议尝试 Envoy。它是子进程的包装器,而子进程又旨在替换旧的模块和函数。Envoy 是人类的子进程。
README 中的示例用法:
>>> r = envoy.run('git config', data='data to pipe in', timeout=2)
>>> r.status_code
129
>>> r.std_out
'usage: git config [options]'
>>> r.std_err
''
管道周围也有东西:
>>> r = envoy.run('uptime | pbcopy')
>>> r.command
'pbcopy'
>>> r.status_code
0
>>> r.history
[<Response 'uptime'>]
使用标准库
使用 subprocess 模块 (Python 3):
import subprocess
subprocess.run(['ls', '-l'])
这是推荐的标准方式。但是,更复杂的任务(管道、输出、输入等)的构造和编写可能很繁琐。
关于 Python 版本的注意事项:如果您仍在使用 Python 2,subprocess.call 的工作方式与此类似。
专业提示:shlex.split 可以帮助您解析 run
、call
和其他子进程
函数的命令,以防您不想(或不能)以列表的形式提供它们:
import shlex
import subprocess
subprocess.run(shlex.split('ls -l'))
具有外部依赖关系
如果您不介意外部依赖,请使用 plumbum:
from plumbum.cmd import ifconfig
print(ifconfig['wlan0']())
它是最好的包装器。它是跨平台的,即它适用于 Windows 和类 Unix 系统。安装方式 。subprocess
pip install plumbum
另一个流行的库是 sh:
from sh import ifconfig
print(ifconfig('wlan0'))
但是,放弃了对 Windows 的支持,因此它不像以前那么棒了。安装方式 。sh
pip install sh
在 Python 中有很多不同的方法可以运行外部命令, 它们都有自己的优点和缺点。
我和我的同事一直在编写 Python 系统管理工具,所以我们需要运行大量的外部命令,有时你希望它们阻塞或异步运行、超时、每秒更新一次等。
还有不同的方法来处理返回代码和错误, 您可能希望解析输出,并提供新的输入(以预期的样式)。或者,您需要重定向标准输入、标准输出和标准错误以在不同的 tty 中运行(例如,使用 GNU Screen 时)。
因此,您可能需要围绕外部命令编写大量包装器。所以这里有一个我们编写的 Python 模块,它可以处理 几乎任何你想要的东西,如果没有,它非常灵活,所以你可以很容易地扩展它:
https://github.com/hpcugent/vsc-base/blob/master/lib/vsc/utils/run.py
它不是独立工作的,需要我们的一些其他工具,并且多年来获得了许多专门的功能,因此它可能不是您的直接替代品,但它可以为您提供大量有关用于运行命令的 Python 内部如何工作的信息以及如何处理某些情况的想法。
2015 年更新:Python 3.5 添加了 subprocess.run,它比 subprocess 更容易使用。噗噗。我推荐。
>>> subprocess.run(["ls", "-l"]) # doesn't capture output
CompletedProcess(args=['ls', '-l'], returncode=0)
>>> subprocess.run("exit 1", shell=True, check=True)
Traceback (most recent call last):
...
subprocess.CalledProcessError: Command 'exit 1' returned non-zero exit status 1
>>> subprocess.run(["ls", "-l", "/dev/null"], capture_output=True)
CompletedProcess(args=['ls', '-l', '/dev/null'], returncode=0,
stdout=b'crw-rw-rw- 1 root root 1, 3 Jan 23 16:23 /dev/null\n', stderr=b'')
评论
只是为了补充讨论,如果包括使用 Python 控制台,则可以从 IPython 调用外部命令。在 IPython 提示符下,可以通过前缀“!”来调用 shell 命令。您还可以将 Python 代码与 shell 结合使用,并将 shell 脚本的输出分配给 Python 变量。
例如:
In [9]: mylist = !ls
In [10]: mylist
Out[10]:
['file1',
'file2',
'file3',]
经过一番研究,我有以下代码,对我来说效果很好。它基本上可以实时打印标准输出和标准误差。
stdout_result = 1
stderr_result = 1
def stdout_thread(pipe):
global stdout_result
while True:
out = pipe.stdout.read(1)
stdout_result = pipe.poll()
if out == '' and stdout_result is not None:
break
if out != '':
sys.stdout.write(out)
sys.stdout.flush()
def stderr_thread(pipe):
global stderr_result
while True:
err = pipe.stderr.read(1)
stderr_result = pipe.poll()
if err == '' and stderr_result is not None:
break
if err != '':
sys.stdout.write(err)
sys.stdout.flush()
def exec_command(command, cwd=None):
if cwd is not None:
print '[' + ' '.join(command) + '] in ' + cwd
else:
print '[' + ' '.join(command) + ']'
p = subprocess.Popen(
command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd
)
out_thread = threading.Thread(name='stdout_thread', target=stdout_thread, args=(p,))
err_thread = threading.Thread(name='stderr_thread', target=stderr_thread, args=(p,))
err_thread.start()
out_thread.start()
out_thread.join()
err_thread.join()
return stdout_result + stderr_result
评论
使用 subprocess.call:
from subprocess import call
# Using list
call(["echo", "Hello", "world"])
# Single string argument varies across platforms so better split it
call("echo Hello world".split(" "))
我倾向于将 subprocess 与 shlex 一起使用(以处理带引号的字符串的转义):
>>> import subprocess, shlex
>>> command = 'ls -l "/your/path/with spaces/"'
>>> call_params = shlex.split(command)
>>> print call_params
["ls", "-l", "/your/path/with spaces/"]
>>> subprocess.call(call_params)
shell.py,我为此写了一个库。
目前,它基本上是 popen 和 shlex 的包装器。它还支持管道命令,因此您可以在 Python 中更轻松地链接命令。因此,您可以执行以下操作:
ex('echo hello shell.py') | "awk '{print $2}'"
一个简单的方法是使用 os 模块:
import os
os.system('ls')
或者,您也可以使用子进程模块:
import subprocess
subprocess.check_call('ls')
如果要将结果存储在变量中,请尝试:
import subprocess
r = subprocess.check_output('ls')
还有 Plumbum
>>> from plumbum import local
>>> ls = local["ls"]
>>> ls
LocalCommand(<LocalPath /bin/ls>)
>>> ls()
u'build.py\ndist\ndocs\nLICENSE\nplumbum\nREADME.rst\nsetup.py\ntests\ntodo.txt\n'
>>> notepad = local["c:\\windows\\notepad.exe"]
>>> notepad() # Notepad window pops up
u'' # Notepad window is closed by user, command returns
评论
用:
import os
cmd = 'ls -al'
os.system(cmd)
os - 此模块提供了一种使用与操作系统相关的功能的可移植方式。
有关更多功能,请参阅文档。os
评论
使用 Python 模块的功能是运行 Linux 命令的最简单方法。在这种情况下,该函数将为您的命令输出。例如Popen
subprocess
Popen.communicate()
import subprocess
..
process = subprocess.Popen(..) # Pass command and arguments to the function
stdout, stderr = process.communicate() # Get command output and error
..
评论
subprocess.check_call()
Popen()
subprocess.run()
这是我的两分钱: 在我看来,这是处理外部命令时的最佳实践......
这些是 execute 方法的返回值...
pass, stdout, stderr = execute(["ls","-la"],"/home/user/desktop")
这是 execute 方法...
def execute(cmdArray,workingDir):
stdout = ''
stderr = ''
try:
try:
process = subprocess.Popen(cmdArray,cwd=workingDir, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=1)
except OSError:
return [False, '', 'ERROR : command(' + ' '.join(cmdArray) + ') could not get executed!']
for line in iter(process.stdout.readline, b''):
try:
echoLine = line.decode("utf-8")
except:
echoLine = str(line)
stdout += echoLine
for line in iter(process.stderr.readline, b''):
try:
echoLine = line.decode("utf-8")
except:
echoLine = str(line)
stderr += echoLine
except (KeyboardInterrupt,SystemExit) as err:
return [False,'',str(err)]
process.stdout.close()
returnCode = process.wait()
if returnCode != 0 or stderr != '':
return [False, stdout, stderr]
else:
return [True, stdout, stderr]
评论
.communicate
Popen()
subprocess.run()
对于 Python 3.5+,建议使用 subprocess 模块中的 run 函数。这将返回一个对象,您可以从中轻松获取输出和返回代码。CompletedProcess
from subprocess import PIPE, run
command = ['echo', 'hello']
result = run(command, stdout=PIPE, stderr=PIPE, universal_newlines=True)
print(result.returncode, result.stdout, result.stderr)
评论
我推荐以下方法“运行”,它将帮助我们以字典的形式获取标准输出、标准错误和退出状态;此方法的调用者可以通过“run”方法读取字典返回,以了解进程的实际状态。
def run (cmd):
print "+ DEBUG exec({0})".format(cmd)
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, shell=True)
(out, err) = p.communicate()
ret = p.wait()
out = filter(None, out.split('\n'))
err = filter(None, err.split('\n'))
ret = True if ret == 0 else False
return dict({'output': out, 'error': err, 'status': ret})
#end
评论
subprocess.run()
shell=True
#end
在 Windows 中,您只需导入模块并通过调用 来运行外部命令,如下所示:subprocess
subprocess.Popen()
subprocess.Popen().communicate()
subprocess.Popen().wait()
# Python script to run a command line
import subprocess
def execute(cmd):
"""
Purpose : To execute a command and return exit status
Argument : cmd - command to execute
Return : exit_code
"""
process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(result, error) = process.communicate()
rc = process.wait()
if rc != 0:
print "Error: failed to execute command:", cmd
print error
return result
# def
command = "tasklist | grep python"
print "This process detail: \n", execute(command)
输出:
This process detail:
python.exe 604 RDP-Tcp#0 4 5,660 K
用:
import subprocess
p = subprocess.Popen("df -h", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).communicate()[0]
print p.split("\n")
它提供了更好的输出,更容易使用:
['Filesystem Size Used Avail Use% Mounted on',
'/dev/sda6 32G 21G 11G 67% /',
'none 4.0K 0 4.0K 0% /sys/fs/cgroup',
'udev 1.9G 4.0K 1.9G 1% /dev',
'tmpfs 387M 1.4M 386M 1% /run',
'none 5.0M 0 5.0M 0% /run/lock',
'none 1.9G 58M 1.9G 3% /run/shm',
'none 100M 32K 100M 1% /run/user',
'/dev/sda5 340G 222G 100G 69% /home',
'']
要从 OpenStack Neutron 获取网络 ID,请执行以下操作:
#!/usr/bin/python
import os
netid = "nova net-list | awk '/ External / { print $2 }'"
temp = os.popen(netid).read() /* Here temp also contains new line (\n) */
networkId = temp.rstrip()
print(networkId)
nova net-list 的输出
+--------------------------------------+------------+------+
| ID | Label | CIDR |
+--------------------------------------+------------+------+
| 431c9014-5b5d-4b51-a357-66020ffbb123 | test1 | None |
| 27a74fcd-37c0-4789-9414-9531b7e3f126 | External | None |
| 5a2712e9-70dc-4b0e-9281-17e02f4684c9 | management | None |
| 7aa697f5-0e60-4c15-b4cc-9cb659698512 | Internal | None |
+--------------------------------------+------------+------+
print(networkId) 的输出
27a74fcd-37c0-4789-9414-9531b7e3f126
评论
os.popen()
调用命令的方法有很多种。
- 例如:
如果需要两个参数。在 cmd 中,我们可以调用 use this:,它会显示在屏幕上。and.exe
sample.exe
and.exe 2 3
5
如果我们使用 Python 脚本来调用,我们应该这样做。and.exe
os.system(cmd,...)
os.system(("and.exe" + " " + "2" + " " + "3"))
os.popen(cmd,...)
os.popen(("and.exe" + " " + "2" + " " + "3"))
subprocess.Popen(cmd,...)
subprocess.Popen(("and.exe" + " " + "2" + " " + "3"))
这太难了,所以我们可以用空格连接cmd:
import os
cmd = " ".join(exename,parameters)
os.popen(cmd)
评论
os.popen
不应该被推荐,甚至可能不再被提及。该示例应将参数作为列表传递,而不是用空格将它们连接起来。subpocess
下面是调用外部命令并返回或打印命令的输出:
Python 子进程 check_output 适用于
运行带有参数的命令,并将其输出作为字节字符串返回。
import subprocess
proc = subprocess.check_output('ipconfig /all')
print proc
评论
shell=True
universal_newlines=True
subproces.run()
有许多不同的库允许您使用 Python 调用外部命令。对于每个库,我都给出了描述,并展示了调用外部命令的示例。我用作示例的命令是(列出所有文件)。如果您想了解有关任何库的更多信息,我列出了每个库并链接了每个库的文档。ls -l
来源
- 子进程:https://docs.python.org/3.5/library/subprocess.html
- shlex:https://docs.python.org/3/library/shlex.html
- 操作系统: https://docs.python.org/3.5/library/os.html
- SH:https://amoffat.github.io/sh/
- Plumbum:https://plumbum.readthedocs.io/en/latest/
- 预期:https://pexpect.readthedocs.io/en/stable/
- 面料: http://www.fabfile.org/
- 特使:https://github.com/kennethreitz/envoy
- 命令:https://docs.python.org/2/library/commands.html
这些都是库
希望这能帮助您决定使用哪个库:)
子流程
子进程允许您调用外部命令并将它们连接到其输入/输出/错误管道(stdin、stdout 和 stderr)。Subprocess 是运行命令的默认选择,但有时其他模块更好。
subprocess.run(["ls", "-l"]) # Run command
subprocess.run(["ls", "-l"], stdout=subprocess.PIPE) # This will run the command and return any output
subprocess.run(shlex.split("ls -l")) # You can also use the shlex library to split the command
操作系统
OS 用于“操作系统相关功能”。它也可以用来调用带有 和 的外部命令(注意:还有一个 subprocess.popen)。操作系统将始终运行 shell,对于不需要或不知道如何使用的人来说,这是一个简单的替代方案。os.system
os.popen
subprocess.run
os.system("ls -l") # Run command
os.popen("ls -l").read() # This will run the command and return any output
SH
sh 是一个子进程接口,它允许您像调用函数一样调用程序。如果要多次运行命令,这将非常有用。
sh.ls("-l") # Run command normally
ls_cmd = sh.Command("ls") # Save command as a variable
ls_cmd() # Run command as if it were a function
铅
plumbum 是一个用于“类似脚本”的 Python 程序的库。您可以调用类似函数的程序,如 中所示。如果您想在没有 shell 的情况下运行管道,Plumbum 非常有用。sh
ls_cmd = plumbum.local("ls -l") # Get command
ls_cmd() # Run command
期望
pexpect 允许您生成子应用程序,控制它们并在其输出中查找模式。对于在 Unix 上期望 tty 的命令,这是 subprocess 的更好替代方案。
pexpect.run("ls -l") # Run command as normal
child = pexpect.spawn('scp foo [email protected]:.') # Spawns child application
child.expect('Password:') # When this is the output
child.sendline('mypassword')
织物
fabric 是一个 Python 2.5 和 2.7 库。它允许您执行本地和远程 shell 命令。Fabric 是在安全外壳 (SSH) 中运行命令的简单替代方案
fabric.operations.local('ls -l') # Run command as normal
fabric.operations.local('ls -l', capture = True) # Run command and receive output
特使
Envoy 被称为“人类的子流程”。它被用作模块周围的便利包装器。subprocess
r = envoy.run("ls -l") # Run command
r.std_out # Get output
命令
commands
包含 的包装函数,但它已从 Python 3 中删除,因为它是更好的选择。os.popen
subprocess
评论
ls_cmd = plumbum.local["ls -l"]
在 Linux 下,如果您想调用将独立执行的外部命令(在 Python 脚本终止后将继续运行),您可以使用简单的队列作为任务后台处理程序或 at 命令。
任务后台处理程序的示例:
import os
os.system('ts <your-command>')
关于任务后台处理程序的注意事项 ():ts
您可以使用以下命令设置要运行的并发进程数(“插槽”):
ts -S <number-of-slots>
安装不需要管理员权限。你可以用一个简单的从源代码下载和编译它,把它添加到你的路径中,你就完成了。
ts
make
评论
ts
在我所知道的任何发行版上都不是标准的,尽管指向的指针有点用。您可能还应该提到.与其他地方一样,该建议可能至少应该提到这是其建议的替代者。at
batch
os.system()
subprocess
如何从 Python 执行程序或调用系统命令
简单,使用 ,它返回一个对象:subprocess.run
CompletedProcess
>>> from subprocess import run
>>> from shlex import split
>>> completed_process = run(split('python --version'))
Python 3.8.8
>>> completed_process
CompletedProcess(args=['python', '--version'], returncode=0)
(run
想要一个词法解析的 shell 参数列表 - 这是您在 shell 中键入的内容,用空格分隔,但不是引号中的空格,因此请使用专用函数 ,将您真正键入的内容拆分到 shell 中)split
为什么?
从 Python 3.5 开始,文档建议使用 subprocess.run:
调用子进程的推荐方法是将 run() 函数用于它可以处理的所有用例。对于更高级的用例,可以直接使用底层 Popen 接口。
下面是一个最简单的用法示例 - 它完全按照要求执行:
>>> from subprocess import run
>>> from shlex import split
>>> completed_process = run(split('python --version'))
Python 3.8.8
>>> completed_process
CompletedProcess(args=['python', '--version'], returncode=0)
run
等待命令成功完成,然后返回一个对象。相反,它可能会引发(如果你给它一个论据)或(如果它失败了,你通过了)。CompletedProcess
TimeoutExpired
timeout=
CalledProcessError
check=True
正如你从上面的例子中可以推断出的那样,默认情况下,stdout 和 stderr 都会通过管道传递到你自己的 stdout 和 stderr。
我们可以检查返回的对象并查看给定的命令和返回码:
>>> completed_process.args
['python', '--version']
>>> completed_process.returncode
0
捕获输出
如果要捕获输出,可以传递给相应的或:subprocess.PIPE
stderr
stdout
>>> from subprocess import PIPE
>>> completed_process = run(shlex.split('python --version'), stdout=PIPE, stderr=PIPE)
>>> completed_process.stdout
b'Python 3.8.8\n'
>>> completed_process.stderr
b''
这些各自的属性返回字节。
传递命令列表
人们可以很容易地从手动提供命令字符串(如问题所建议的那样)转变为提供以编程方式构建的字符串。不要以编程方式生成字符串。这是一个潜在的安全问题。最好假设你不信任输入。
>>> import textwrap
>>> args = ['python', textwrap.__file__]
>>> cp = run(args, stdout=subprocess.PIPE)
>>> cp.stdout
b'Hello there.\n This is indented.\n'
请注意,仅应按位置传递。args
完整签名
这是源代码中的实际签名,如下所示:help(run)
def run(*popenargs, input=None, timeout=None, check=False, **kwargs):
和 被赋予构造函数。 可以是将通过管道传递到子进程的 stdin 的字节字符串(如果指定编码或 ),也可以是 unicode。popenargs
kwargs
Popen
input
universal_newlines=True
文档描述并且比我做得更好:timeout=
check=True
timeout 参数被传递给 Popen.communicate()。如果超时 过期时,子进程将被终止并等待。这 TimeoutExpired 异常将在子进程 终止。
如果 check 为 true,并且进程以非零退出代码退出,则 将引发 CalledProcessError 异常。该属性 异常包含参数、退出代码以及 stdout 和 stderr if 他们被俘虏了。
这个例子比我能想出的更好:check=True
>>> subprocess.run("exit 1", shell=True, check=True) Traceback (most recent call last): ... subprocess.CalledProcessError: Command 'exit 1' returned non-zero exit status 1
扩展签名
下面是一个扩展的签名,如文档中所示:
subprocess.run(args, *, stdin=None, input=None, stdout=None, stderr=None, shell=False, cwd=None, timeout=None, check=False, encoding=None, errors=None)
请注意,这表示只应按位置传递 args 列表。因此,将其余参数作为关键字参数传递。
普彭
什么时候改用?我很难仅根据论点找到用例。但是,直接使用 would,您可以访问其方法,包括 、'send_signal'、'terminate' 和 'wait'。Popen
Popen
poll
这是源中给出的签名。我认为这是对信息最精确的封装(而不是):Popen
help(Popen)
def __init__(self, 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=(), *, user=None, group=None, extra_groups=None,
encoding=None, errors=None, text=None, umask=-1, pipesize=-1):
但信息量更大的是 Popen
文档:
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=(), *, group=None, extra_groups=None, user=None, umask=-1, encoding=None, errors=None, text=None)
在新进程中执行子程序。在 POSIX 上,该类使用 os.execvp() 类行为来执行子程序。在 Windows 上, 该类使用 Windows CreateProcess() 函数。参数 Popen如下。
了解其余的文档将留给读者作为练习。Popen
评论
我写了一个包装器来处理错误和重定向输出和其他东西。
import shlex
import psutil
import subprocess
def call_cmd(cmd, stdout=sys.stdout, quiet=False, shell=False, raise_exceptions=True, use_shlex=True, timeout=None):
"""Exec command by command line like 'ln -ls "/var/log"'
"""
if not quiet:
print("Run %s", str(cmd))
if use_shlex and isinstance(cmd, (str, unicode)):
cmd = shlex.split(cmd)
if timeout is None:
process = subprocess.Popen(cmd, stdout=stdout, stderr=sys.stderr, shell=shell)
retcode = process.wait()
else:
process = subprocess.Popen(cmd, stdout=stdout, stderr=sys.stderr, shell=shell)
p = psutil.Process(process.pid)
finish, alive = psutil.wait_procs([p], timeout)
if len(alive) > 0:
ps = p.children()
ps.insert(0, p)
print('waiting for timeout again due to child process check')
finish, alive = psutil.wait_procs(ps, 0)
if len(alive) > 0:
print('process {} will be killed'.format([p.pid for p in alive]))
for p in alive:
p.kill()
if raise_exceptions:
print('External program timeout at {} {}'.format(timeout, cmd))
raise CalledProcessTimeout(1, cmd)
retcode = process.wait()
if retcode and raise_exceptions:
print("External program failed %s", str(cmd))
raise subprocess.CalledProcessError(retcode, cmd)
你可以这样称呼它:
cmd = 'ln -ls "/var/log"'
stdout = 'out.txt'
call_cmd(cmd, stdout)
举个例子(在 Linux 中):
import subprocess
subprocess.run('mkdir test.dir', shell=True)
这将在当前目录中创建 test.dir。 请注意,这也有效:
import subprocess
subprocess.call('mkdir test.dir', shell=True)
使用 os.system 的等效代码为:
import os
os.system('mkdir test.dir')
最佳做法是使用 subprocess 而不是 os,使用 .run 而不是 .call。 您需要了解的有关子流程的所有信息都在这里。 另外,请注意,所有 Python 文档都可以从这里下载。我下载了打包为 .zip 的 PDF。我之所以提到这一点,是因为教程.pdf(第 81 页)中有一个很好的 os 模块概述。此外,它是 Python 编码人员的权威资源。
评论
通常,我将以下函数用于外部命令,这对于长时间运行的进程特别方便。以下方法在进程运行时跟踪进程输出并返回输出,如果进程失败,则引发异常。
如果在进程上使用 poll() 方法完成进程,它就会出现。
import subprocess,sys
def exec_long_running_proc(command, args):
cmd = "{} {}".format(command, " ".join(str(arg) if ' ' not in arg else arg.replace(' ','\ ') for arg in args))
print(cmd)
process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
# Poll process for new output until finished
while True:
nextline = process.stdout.readline().decode('UTF-8')
if nextline == '' and process.poll() is not None:
break
sys.stdout.write(nextline)
sys.stdout.flush()
output = process.communicate()[0]
exitCode = process.returncode
if (exitCode == 0):
return output
else:
raise Exception(command, exitCode, output)
你可以像这样调用它:
exec_long_running_proc(command = "hive", args=["-f", hql_path])
评论
repr(arg)
str(arg)
repr(arg)
exec_long_running_proc(command = "ls", args=["-l", "~/test file*"])
在 Python 中调用外部命令
调用外部命令的一种简单方法是使用 。此函数返回命令的退出值。但缺点是我们不会得到 stdout 和 stderr。os.system(...)
ret = os.system('some_cmd.sh')
if ret != 0 :
print 'some_cmd.sh execution returned failure'
在后台调用 Python 中的外部命令
subprocess.Popen
为运行外部命令提供了更大的灵活性,而不是使用 。我们可以在后台启动一个命令并等待它完成。之后,我们可以得到 stdout 和 stderr。os.system
proc = subprocess.Popen(["./some_cmd.sh"], stdout=subprocess.PIPE)
print 'waiting for ' + str(proc.pid)
proc.wait()
print 'some_cmd.sh execution finished'
(out, err) = proc.communicate()
print 'some_cmd.sh output : ' + out
在后台调用 Python 中长时间运行的外部命令并在一段时间后停止
我们甚至可以在后台启动一个长时间运行的进程,并在其任务完成后的某个时间后将其杀死。subprocess.Popen
proc = subprocess.Popen(["./some_long_run_cmd.sh"], stdout=subprocess.PIPE)
# Do something else
# Now some_long_run_cmd.sh exeuction is no longer needed, so kill it
os.system('kill -15 ' + str(proc.pid))
print 'Output : ' proc.communicate()[0]
它可以是这么简单:
import os
cmd = "your command"
os.system(cmd)
评论
如果需要从 Python 笔记本(如 Jupyter、Zeppelin、Databricks 或 Google Cloud Datalab)调用 shell 命令,只需使用前缀即可。!
例如
!ls -ilF
Invoke 是一个 Python(2.7 和 3.4+)任务执行工具和库。它提供了一个干净的高级 API,用于运行 shell 命令:
>>> from invoke import run
>>> cmd = "pip install -r requirements.txt"
>>> result = run(cmd, hide=True, warn=True)
>>> print(result.ok)
True
>>> print(result.stdout.splitlines()[-1])
Successfully installed invocations-0.13.0 pep8-1.5.7 spec-1.3.1
评论
invoke
subprocess
requests
urllib3
我写了一个小库来帮助这个用例:
https://pypi.org/project/citizenshell/
可以使用以下方法进行安装
pip install citizenshell
然后按如下方式使用:
from citizenshell import sh
assert sh("echo Hello World") == "Hello World"
您可以将标准输出与标准错误分开,并提取退出代码,如下所示:
result = sh(">&2 echo error && echo output && exit 13")
assert result.stdout() == ["output"]
assert result.stderr() == ["error"]
assert result.exit_code() == 13
很酷的是,在开始处理输出之前,您不必等待底层 shell 退出:
for line in sh("for i in 1 2 3 4; do echo -n 'It is '; date +%H:%M:%S; sleep 1; done", wait=False)
print ">>>", line + "!"
由于 wait=False,将在可用时打印行
>>> It is 14:24:52!
>>> It is 14:24:53!
>>> It is 14:24:54!
>>> It is 14:24:55!
更多示例可在 https://github.com/meuter/citizenshell 上找到
为了在 Python 3.5+ 中使用,以下方法在 Linux 上为我提供了帮助:subprocess
import subprocess
# subprocess.run() returns a completed process object that can be inspected
c = subprocess.run(["ls", "-ltrh"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
print(c.stdout.decode('utf-8'))
如文档中所述,值是字节序列,为了正确显示它们,应考虑解码。对于更高版本的 Python,并添加到 subprocess.run()
的 kwargs 中。PIPE
text=True
encoding='utf-8'
上述代码的输出为:
total 113M
-rwxr-xr-x 1 farzad farzad 307 Jan 15 2018 vpnscript
-rwxrwxr-x 1 farzad farzad 204 Jan 15 2018 ex
drwxrwxr-x 4 farzad farzad 4.0K Jan 22 2018 scripts
.... # Some other lines
如果未在命令中使用用户输入,则可以使用以下命令:
from os import getcwd
from subprocess import check_output
from shlex import quote
def sh(command):
return check_output(quote(command), shell=True, cwd=getcwd(), universal_newlines=True).strip()
并把它用作
branch = sh('git rev-parse --abbrev-ref HEAD')
shell=True
会生成一个 shell,所以你可以用 pipe 之类的 shell 东西。这对于运行硬编码命令和处理其输出非常非常方便。确保输出以字符串而不是二进制形式返回。sh('ps aux | grep python')
universal_lines=True
cwd=getcwd()
将确保使用与解释器相同的工作目录运行该命令。这对于 Git 命令来说很方便,可以像上面的 Git 分支名称示例一样工作。
一些食谱
- 可用内存(以兆字节为单位):
sh('free -m').split('\n')[1].split()[1]
- 可用空间开/以百分比表示
sh('df -m /').split('\n')[1].split()[4][0:-1]
- CPU 负载
sum(map(float, sh('ps -ef -o pcpu').split('\n')[1:])
但这对于用户输入来说并不安全,从文档中可以看出:
安全注意事项
与其他一些 popen 函数不同,此实现永远不会 隐式调用系统 shell。这意味着所有字符, 包括 shell 元字符,可以安全地传递给子项 过程。如果通过 shell=True 显式调用 shell,则 应用程序有责任确保所有空格和 适当引用元字符以避免 shell 注入 漏洞。
使用 shell=True 时,shlex.quote() 函数可用于 正确转义字符串中的空格和 shell 元字符 将用于构造 shell 命令。
即使使用 ,在 shell 命令上使用用户输入时,最好保持一点偏执。一种选择是使用硬编码命令来获取一些通用输出并按用户输入进行过滤。无论如何,使用将确保只执行您要执行的确切进程,否则会出现错误。shlex.quote()
shell=False
No such file or directory
此外,根据我的测试,它似乎比(默认值)慢了 20% 左右。shell=True
shell=False
In [50]: timeit("check_output('ls -l'.split(), universal_newlines=True)", number=1000, globals=globals())
Out[50]: 2.6801227919995654
In [51]: timeit("check_output('ls -l', universal_newlines=True, shell=True)", number=1000, globals=globals())
Out[51]: 3.243950183999914
Sultan 是最近用于此目的的一揽子计划。它提供了一些有关管理用户权限和添加有用错误消息的细节。
from sultan.api import Sultan
with Sultan.load(sudo=True, hostname="myserver.com") as sultan:
sultan.yum("install -y tree").run()
如果要编写 Python shell 脚本并在系统上安装了 IPython,则可以使用 bang 前缀在 IPython 中运行 shell 命令:
!ls
filelist = !ls
评论
蟒蛇 3.5+
import subprocess
p = subprocess.run(["ls", "-ltr"], capture_output=True)
print(p.stdout.decode(), p.stderr.decode())
import subprocess
p = subprocess.run(["ls", "-ltr"], capture_output=True)
print(p.stdout.decode(), p.stderr.decode())
评论
os.popen()
是执行命令的最简单、最安全的方式。您可以执行在命令行上运行的任何命令。此外,您还可以使用os.popen().read()
你可以这样做:
import os
output = os.popen('Your Command Here').read()
print (output)
列出当前目录中所有文件的示例:
import os
output = os.popen('ls').read()
print (output)
# Outputs list of files in the directory
大多数情况下:
在大多数情况下,您只需要一小段代码:
import subprocess
import shlex
source = "test.txt"
destination = "test_copy.txt"
base = "cp {source} {destination}'"
cmd = base.format(source=source, destination=destination)
subprocess.check_call(shlex.split(cmd))
它干净而简单。
subprocess.check_call
运行带有参数的命令并等待 命令完成。
shlex.split
使用类似 shell 的语法拆分字符串 cmd
其余案例:
如果这不适用于某些特定命令,则很可能是命令行解释器有问题。操作系统选择了不适合您的程序类型或在系统可执行路径上找不到足够的程序的默认程序。
例:
在 Unix 系统上使用重定向运算符
input_1 = "input_1.txt"
input_2 = "input_2.txt"
output = "merged.txt"
base_command = "/bin/bash -c 'cat {input} >> {output}'"
base_command.format(input_1, output=output)
subprocess.check_call(shlex.split(base_command))
base_command.format(input_2, output=output)
subprocess.check_call(shlex.split(base_command))
正如《Python 的禅宗》中所说:显式比 含蓄
因此,如果使用 Python >=3.6 函数,它将如下所示:
import subprocess
import shlex
def run_command(cmd_interpreter: str, command: str) -> None:
base_command = f"{cmd_interpreter} -c '{command}'"
subprocess.check_call(shlex.split(base_command)
有多种方法可以从 Python 调用外部命令。有一些函数和模块带有好帮手功能,可以使它变得非常容易。但推荐的是模块。subprocess
import subprocess as s
s.call(["command.exe", "..."])
调用函数将启动外部进程,传递一些命令行参数并等待它完成。完成后,继续执行。调用函数中的参数通过列表传递。列表中的第一个参数是命令,通常采用可执行文件的形式,列表中的后续参数是要传递的任何参数。
如果您以前在窗口中从命令行调用过进程,您将意识到您经常需要引用参数。您需要在它周围加上引号。如果有一个空格,那么就会有一个反斜杠,并且有一些复杂的规则,但是你可以通过使用模块在 Python 中避免很多这种情况,因为它是一个列表,每个项目都是不同的,python 可以为你正确引用。subprocess
最后,在列表之后,有许多可选参数,其中一个是 shell,如果将 shell equals 设置为 true,则您的命令将像在命令提示符下键入一样运行。
s.call(["command.exe", "..."], shell=True)
这使您可以访问管道等功能,可以重定向到文件,可以在一件事中调用多个命令。
还有一件事,如果你的脚本依赖于进程成功,那么你要检查结果,并且可以使用检查调用帮助程序函数检查结果。
s.check_call(...)
它与调用函数完全相同,它接受相同的参数,接受相同的列表,您可以传入任何额外的参数,但它将等待函数完成。如果函数的退出代码不是零,它将通过 python 脚本中的异常。
最后,如果你想要更严格的控制构造函数,它也来自模块。它也采用与 incall 和 check_call 函数相同的参数,但它返回一个表示正在运行的进程的对象。Popen
subprocess
p=s.Popen("...")
它不会等待正在运行的进程完成,也不会立即抛出任何异常,但它会给你一个对象,让你做一些事情,比如等待它完成,让你与它通信,你可以重定向标准输入,标准输出如果你想在其他地方显示输出等等。
我将其用于 Python 3.6+:
import subprocess
def execute(cmd):
"""
Purpose : To execute a command and return exit status
Argument : cmd - command to execute
Return : result, exit_code
"""
process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(result, error) = process.communicate()
rc = process.wait()
if rc != 0:
print ("Error: failed to execute command: ", cmd)
print (error.rstrip().decode("utf-8"))
return result.rstrip().decode("utf-8"), serror.rstrip().decode("utf-8")
# def
评论
shell=True
cmd=["/bin/echo", "hello word"]
从 2018 年 6 月 27 日发布的 Python 3.7.0 开始(https://docs.python.org/3/whatsnew/3.7.html),您可以以最强大但同样简单的方式实现所需的结果。本答案旨在以简短的方式向您展示各种选项的基本摘要。有关深入的答案,请参阅其他答案。
TL;DR 在2021之内
最大的优点是它的简单性。 更好,仍然易于使用,尤其是从 Python 3.5 开始。os.system(...)
subprocess
import subprocess
subprocess.run("ls -a", shell=True)
注意:这是您的问题的确切答案 - 运行命令
就像在贝壳中一样
首选方式
如果可能,请删除 shell 开销并直接运行命令(需要列表)。
import subprocess
subprocess.run(["help"])
subprocess.run(["ls", "-a"])
在列表中传递程序参数。对于包含空格的参数,不要包含 \“
-转义。
高级用例
检查输出
以下代码不言自明:
import subprocess
result = subprocess.run(["ls", "-a"], capture_output=True, text=True)
if "stackoverflow-logo.png" in result.stdout:
print("You're a fan!")
else:
print("You're not a fan?")
result.stdout
是除错误外的所有正常程序输出。阅读以获取它们。result.stderr
capture_output=True
- 开启拍摄。否则,将是.从 Python 3.7 开始可用。result.stderr
result.stdout
None
text=True
- Python 3.7 中添加了一个方便参数,用于将接收到的二进制数据转换为可以轻松使用的 Python 字符串。
检查返回码
做
if result.returncode == 127: print("The program failed for some weird reason")
elif result.returncode == 0: print("The program succeeded")
else: print("The program failed unexpectedly")
如果你只是想检查程序是否成功(returncode == 0)并抛出异常,有一个更方便的功能:
result.check_returncode()
但它是 Python,所以有一个更方便的论点,它会自动为你做同样的事情:check
result = subprocess.run(..., check=True)
stderr 应该在 stdout 中
您可能希望所有程序输出都在 stdout 中,甚至错误。为此,请运行
result = subprocess.run(..., stderr=subprocess.STDOUT)
result.stderr
然后将包含所有内容。None
result.stdout
将 shell=False 与参数字符串一起使用
shell=False
需要参数列表。但是,您可以使用 shlex 自行拆分参数字符串。
import subprocess
import shlex
subprocess.run(shlex.split("ls -a"))
就是这样。
常见问题
当你遇到这个问题时,你很有可能刚开始使用 Python。让我们来看看一些常见问题。
FileNotFoundError: [Errno 2] 没有这样的文件或目录: 'ls -a': 'ls -a'
您正在运行一个没有 shell=True
的子进程。使用列表 ([“ls”, “-a”]
) 或设置 shell=True
。
类型错误:[...]无类型 [...]
检查是否设置了 capture_output=True
。
TypeError:需要类似字节的对象,而不是 [...]
您始终会收到来自程序的字节结果。如果要像普通字符串一样使用它,请设置 text=True
。
子进程。CalledProcessError:命令“[...]”返回非零退出状态 1。
您的命令未成功运行。您可以禁用返回码检查或检查实际程序的有效性。
TypeError:init() 有一个意外的关键字参数 [...]
您可能使用的 Python 版本低于 3.7.0;将其更新为最新的可用版本。否则,这篇 Stack Overflow 帖子中还有其他答案,向您展示了较旧的替代解决方案。
评论
shell=True
os.system(...)
shell=True
Preferred Way
os.system(...)
shell=True
"
os.system(...)
shell=True
您可以使用 subprocess 模块中的 Popen 运行任何命令。
from subprocess import Popen
首先,使用要运行的所有参数创建一个命令对象。例如,在下面的代码片段中,gunicorm 命令对象已由所有参数组成:
cmd = (
"gunicorn "
"-c gunicorn_conf.py "
"-w {workers} "
"--timeout {timeout} "
"-b {address}:{port} "
"--limit-request-line 0 "
"--limit-request-field_size 0 "
"--log-level debug "
"--max-requests {max_requests} "
"manage:app").format(**locals())
然后,此命令对象与 Popen 一起使用以实例化进程:
process = Popen(cmd, shell=True)
也可以根据任何信号终止此过程,使用下面的代码行:
Popen.terminate(process)
您可以等到上述命令的执行完成:
process.wait()
这里有很多答案,但没有一个能满足我的所有需求。
- 我需要运行命令并捕获输出和退出代码。
- 我需要使已执行的程序超时,并在达到超时时强制它退出,并终止其所有子进程。
- 我需要它在 Windows XP 及更高版本、Cygwin 和 Linux 中工作。在 Python 2 和 3 中。
所以我创建了这个:
def _run(command, timeout_s=False, shell=False):
### run a process, capture the output and wait for it to finish. if timeout is specified then Kill the subprocess and its children when the timeout is reached (if parent did not detach)
## usage: _run(arg1, arg2, arg3)
# arg1: command + arguments. Always pass a string; the function will split it when needed
# arg2: (optional) timeout in seconds before force killing
# arg3: (optional) shell usage. default shell=False
## return: a list containing: exit code, output, and if timeout was reached or not
# - Tested on Python 2 and 3 on Windows XP, Windows 7, Cygwin and Linux.
# - preexec_fn=os.setsid (py2) is equivalent to start_new_session (py3) (works on Linux only), in Windows and Cygwin we use TASKKILL
# - we use stderr=subprocess.STDOUT to merge standard error and standard output
import sys, subprocess, os, signal, shlex, time
def _runPY3(command, timeout_s=None, shell=False):
# py3.3+ because: timeout was added to communicate() in py3.3.
new_session=False
if sys.platform.startswith('linux'): new_session=True
p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, start_new_session=new_session, shell=shell)
try:
out = p.communicate(timeout=timeout_s)[0].decode('utf-8')
is_timeout_reached = False
except subprocess.TimeoutExpired:
print('Timeout reached: Killing the whole process group...')
killAll(p.pid)
out = p.communicate()[0].decode('utf-8')
is_timeout_reached = True
return p.returncode, out, is_timeout_reached
def _runPY2(command, timeout_s=0, shell=False):
preexec=None
if sys.platform.startswith('linux'): preexec=os.setsid
p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, preexec_fn=preexec, shell=shell)
start_time = time.time()
is_timeout_reached = False
while timeout_s and p.poll() == None:
if time.time()-start_time >= timeout_s:
print('Timeout reached: Killing the whole process group...')
killAll(p.pid)
is_timeout_reached = True
break
time.sleep(1)
out = p.communicate()[0].decode('utf-8')
return p.returncode, out, is_timeout_reached
def killAll(ParentPid):
if sys.platform.startswith('linux'):
os.killpg(os.getpgid(ParentPid), signal.SIGTERM)
elif sys.platform.startswith('cygwin'):
# subprocess.Popen(shlex.split('bash -c "TASKKILL /F /PID $(</proc/{pid}/winpid) /T"'.format(pid=ParentPid)))
winpid=int(open("/proc/{pid}/winpid".format(pid=ParentPid)).read())
subprocess.Popen(['TASKKILL', '/F', '/PID', str(winpid), '/T'])
elif sys.platform.startswith('win32'):
subprocess.Popen(['TASKKILL', '/F', '/PID', str(ParentPid), '/T'])
# - In Windows, we never need to split the command, but in Cygwin and Linux we need to split if shell=False (default), shlex will split the command for us
if shell==False and (sys.platform.startswith('cygwin') or sys.platform.startswith('linux')):
command=shlex.split(command)
if sys.version_info >= (3, 3): # py3.3+
if timeout_s==False:
returnCode, output, is_timeout_reached = _runPY3(command, timeout_s=None, shell=shell)
else:
returnCode, output, is_timeout_reached = _runPY3(command, timeout_s=timeout_s, shell=shell)
else: # Python 2 and up to 3.2
if timeout_s==False:
returnCode, output, is_timeout_reached = _runPY2(command, timeout_s=0, shell=shell)
else:
returnCode, output, is_timeout_reached = _runPY2(command, timeout_s=timeout_s, shell=shell)
return returnCode, output, is_timeout_reached
然后像这样使用它:
始终将命令作为一个字符串传递(这样更容易)。您不需要拆分它;该函数将在需要时拆分它。
如果你的命令在你的 shell 中有效,它将与这个函数一起使用,所以先在你的 shell 中测试你的命令 cmd/Bash。
因此,我们可以像这样在超时的情况下使用它:
a=_run('cmd /c echo 11111 & echo 22222 & calc',3)
for i in a[1].splitlines(): print(i)
或者没有超时:
b=_run('cmd /c echo 11111 & echo 22222 & calc')
更多示例:
b=_run('''wmic nic where 'NetConnectionID="Local Area Connection"' get NetConnectionStatus /value''')
print(b)
c=_run('cmd /C netsh interface ip show address "Local Area Connection"')
print(c)
d=_run('printf "<%s>\n" "{foo}"')
print(d)
您也可以指定 shell=True,但在大多数情况下,此函数是无用的。我更喜欢自己选择我想要的外壳,但如果你也需要它,这里是:
# windows
e=_run('echo 11111 & echo 22222 & calc',3, shell=True)
print(e)
# Cygwin/Linux:
f=_run('printf "<%s>\n" "{foo}"', shell=True)
print(f)
为什么我没有使用更简单的新方法subprocess.run()
?
- 因为它在 Python 3.7+ 中受支持,但 Windows XP 中支持的最后一个 Python 版本是 3.4。
- 并且由于此函数的 timeout 参数在 Windows 中无用,因此它不会终止已执行命令的子进程。
- 如果使用 + 参数,则如果子进程仍在运行,它将挂起。它在 Windows 中仍然损坏,问题 31447 仍然悬而未决。
capture_output
timeout
评论
subprocess.run()
您可以尝试使用运行外部命令。os.system()
例:
import os
try:
os.system('ls')
pass
except:
print("Error running command")
pass
在此示例中,该脚本导入并尝试运行 中列出的命令。如果命令失败,则它将打印“运行命令时出错”,而不会因错误而停止脚本。os
os.system()
是的,就是这么简单!
下面是一个 Python 脚本,它将在 Ubuntu 上运行该命令,同时还实时显示日志:
command = 'your command here'
process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
while True:
output = process.stdout.readline().decode()
if output == '' and process.poll() is not None:
break
if output:
print(output.strip())
rc = process.poll()
if rc == 0:
print("Command succeeded.")
else:
print("Command failed.")
使用 Python subprocess 模块执行 shell 命令并将输出写入文件。
以下脚本将运行 ps -ef 命令,过滤包含 python3 的行,并将它们写入名为 python_processes.txt 的文件中。请注意,该代码不处理执行过程中可能发生的任何异常。
import subprocess
# Command to execute
cmd = ["ps", "-ef"]
# Execute the command
process = subprocess.Popen(cmd, stdout=subprocess.PIPE)
output, error = process.communicate()
# Check if the command was executed without errors
if error is None:
# Filter lines with 'python3'
python_processes = [line for line in output.decode('utf-8').split('\n') if 'python3' in line]
# Write the output to a file
with open('python_processes.txt', 'w') as f:
for process in python_processes:
f.write(process + '\n')
else:
print(f"Error occurred while executing command: {error}")
评论