5622

如何从 Python 脚本中调用外部命令(就像我在 Unix shell 或 Windows 命令提示符下输入的一样)?

4

63 回答 63

5374

使用subprocess标准库中的模块:

import subprocess
subprocess.run(["ls", "-l"])

subprocess.runover的优点os.system是更灵活(你可以获得stdout, stderr, “真正的”状态码, 更好的错误处理等等...)。

甚至推荐使用的文档os.systemsubprocess

subprocess模块提供了更强大的工具来生成新进程并检索其结果;使用该模块优于使用此功能。有关一些有用的秘诀,请参阅文档中的用子流程模块替换旧功能部分。subprocess

在 Python 3.4 及更早版本上,使用subprocess.call代替.run

subprocess.call(["ls", "-l"])
于 2008-09-18T01:39:35.257 回答
3275

调用外部程序的方法总结,包括它们的优缺点:

  1. os.system将命令和参数传递给系统的 shell。这很好,因为您实际上可以以这种方式一次运行多个命令并设置管道和输入/输出重定向。例如:

    os.system("some_command < input_file | another_command > output_file")  
    

    然而,虽然这很方便,但您必须手动处理 shell 字符的转义,例如空格等。另一方面,这也允许您运行只是 shell 命令而不是实际外部程序的命令。

  2. os.popenos.system除了它给你一个类似文件的对象,你可以使用它来访问该进程的标准输入/输出之外,它将做同样的事情。popen 还有 3 个其他变体,它们对 i/o 的处理都略有不同。如果您将所有内容都作为字符串传递,那么您的命令将传递给 shell;如果您将它们作为列表传递,那么您无需担心转义任何内容。例子:

    print(os.popen("ls -l").read())
    
  3. subprocess.Popen. 这旨在替代os.popen,但由于如此全面,它的缺点是稍微复杂一些。例如,你会说:

    print subprocess.Popen("echo Hello World", shell=True, stdout=subprocess.PIPE).stdout.read()
    

    代替

    print os.popen("echo Hello World").read()
    

    但是很高兴在一个统一的类中拥有所有选项,而不是 4 个不同的 popen 函数。请参阅文档

  4. subprocess.call. 这基本上就像Popen类一样,接受所有相同的参数,但它只是等到命令完成并为您提供返回码。例如:

    return_code = subprocess.call("echo Hello World", shell=True)
    
  5. subprocess.run. 仅限 Python 3.5+。与上面类似,但更灵活,并CompletedProcess在命令执行完成时返回一个对象。

  6. os.fork, os.exec,os.spawn与它们的 C 语言对应物相似,但我不建议直接使用它们。

subprocess模块可能应该是您使用的。

最后,请注意,对于将最终命令作为字符串传递给 shell 的所有方法,您有责任将其转义。如果您传递的字符串的任何部分都不能完全信任,则会产生严重的安全隐患。例如,如果用户正在输入字符串的某些/任何部分。如果您不确定,请仅将这些方法与常量一起使用。为了给您暗示含义,请考虑以下代码:

print subprocess.Popen("echo %s " % user_input, stdout=PIPE).stdout.read()

并想象用户输入了一些“ my mama didnt love me && rm -rf /”,这可能会擦除整个文件系统。

于 2008-09-18T13:11:46.980 回答
413

典型实现:

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().

于 2008-09-18T18:20:46.160 回答
265

关于将子进程与调用进程分离的一些提示(在后台启动子进程)。

假设您想从 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 社区中,问题是一样的。

解决方案是将 DETACHED_PROCESS进程创建标志传递给 Windows API 中的底层 CreateProcess 函数。如果你碰巧安装了pywin32,你可以从win32process模块​​导入flag,否则你应该自己定义:

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 中启动后台进程还没有任何启示。

于 2010-02-12T10:15:34.027 回答
193
import os
os.system("your command")

请注意,这是危险的,因为该命令没有被清除。我把它留给你去谷歌搜索'os'和'sys'模块的相关文档。有很多函数(exec* 和 spawn*)可以做类似的事情。

于 2008-09-18T01:37:24.717 回答
168

我建议使用subprocess模块而不是 os.system ,因为它会为您进行 shell 转义,因此更安全。

subprocess.call(['ping', 'localhost'])
于 2008-09-18T01:42:30.540 回答
163
import os
cmd = 'ls -al'
os.system(cmd)

如果要返回命令的结果,可以使用os.popen. 但是,自 2.6 版起已弃用此功能,转而支持subprocess module,其他答案已很好地涵盖了该功能。

于 2008-09-18T01:37:49.613 回答
125

有许多不同的库允许您使用 Python 调用外部命令。对于每个库,我都给出了描述并展示了调用外部命令的示例。我用作示例的命令是ls -l(列出所有文件)。如果您想了解有关我列出的任何库的更多信息,并链接了每个库的文档。

来源

这些都是图书馆

希望这将帮助您决定使用哪个库:)

子进程

子进程允许您调用外部命令并将它们连接到它们的输入/输出/错误管道(stdin、stdout 和 stderr)。子进程是运行命令的默认选择,但有时其他模块更好。

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 用于“操作系统相关功能”。也可以用os.systemand调用外部命令os.popen(注意:还有一个subprocess.popen)。os 将始终运行 shell,对于不需要或不知道如何使用subprocess.run.

os.system("ls -l") # Run command
os.popen("ls -l").read() # This will run the command and return any output

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 程序的库。您可以调用类似函数的程序,如sh. 如果您想在没有外壳的情况下运行管道,Plumbum 很有用。

ls_cmd = plumbum.local("ls -l") # Get command
ls_cmd() # Run command

期待

pexpect 允许您生成子应用程序、控制它们并在它们的输出中查找模式。对于需要 Unix 上的 tty 的命令,这是子进程的更好替代方案。

pexpect.run("ls -l") # Run command as normal
child = pexpect.spawn('scp foo user@example.com:.') # 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包含 的包装函数os.popen,但它已从 Python 3 中删除,因为subprocess它是一个更好的选择。

于 2016-10-29T14:02:50.753 回答
85

使用标准库

使用子进程模块(Python 3):

import subprocess
subprocess.run(['ls', '-l'])

这是推荐的标准方式。然而,更复杂的任务(管道、输出、输入等)的构建和编写可能会很乏味。

关于 Python 版本的注意事项:如果您仍在使用 Python 2,则subprocess.call以类似的方式工作。

专业提示:shlex.split可以帮助您解析、 和其他函数的命令run,以防您不希望(或者您不能!)以列表的形式提供它们:callsubprocess

import shlex
import subprocess
subprocess.run(shlex.split('ls -l'))

有外部依赖

如果您不介意外部依赖项,请使用plumbum

from plumbum.cmd import ifconfig
print(ifconfig['wlan0']())

这是最好的subprocess包装。它是跨平台的,即它适用于 Windows 和类 Unix 系统。通过安装pip install plumbum

另一个流行的库是sh

from sh import ifconfig
print(ifconfig('wlan0'))

然而,sh放弃了对 Windows 的支持,所以它不像以前那么棒了。通过安装pip install sh

于 2013-04-11T17:17:53.443 回答
83

我总是用fabric这样的东西:

from fabric.operations import local
result = local('ls', capture=True)
print "Content:/n%s" % (result, )

但这似乎是一个很好的工具:sh(Python subprocess interface)

看一个例子:

from sh import vgdisplay
print vgdisplay()
print vgdisplay('-v')
print vgdisplay(v=True)
于 2012-03-13T00:12:54.427 回答
81

检查“pexpect”Python 库。

它允许对外部程序/命令,甚至 ssh、ftp、telnet 等进行交互式控制。您可以输入如下内容:

child = pexpect.spawn('ftp 192.168.0.24')

child.expect('(?i)name .*: ')

child.sendline('anonymous')

child.expect('(?i)password')
于 2010-10-07T07:09:04.150 回答
79

如果您需要正在调用的命令的输出,则可以使用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 是True,指定的命令将通过 shell 执行。如果您将 Python 主要用于它在大多数系统 shell 上提供的增强控制流,并且仍然希望方便地访问其他 shell 功能,例如 shell 管道、文件名通配符、环境变量扩展和 ~ 到用户家的扩展,这将很有用目录。但是,请注意 Python 本身提供了许多类似 shell 的功能的实现(特别是 、、globfnmatchos.walk()和)。os.path.expandvars()os.path.expanduser()shutil

于 2011-04-28T20:29:29.967 回答
66

更新:

subprocess.run如果您的代码不需要保持与早期 Python 版本的兼容性,则从Python 3.5 开始推荐使用此方法。它更加一致,并提供与 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'>]
于 2012-11-15T17:13:22.540 回答
64

这就是我运行命令的方式。这段代码有你需要的一切

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()
于 2012-10-28T05:14:01.820 回答
54

如何从 Python 执行程序或调用系统命令

简单, use 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 中键入的内容,用空格分隔,但不是空格被引用的地方,所以使用一个专门的函数 ,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给相应的stderror 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):

popenargskwargs赋予Popen构造函数。input可以是字节串(或 unicode,如果指定编码或universal_newlines=True),将通过管道传送到子进程的标准输入。

该文档描述timeout=check=True比我更好:

超时参数被传递给 Popen.communicate()。如果超时,子进程将被杀死并等待。TimeoutExpired 异常将在子进程终止后重新引发。

如果检查为真,并且进程以非零退出代码退出,则会引发 CalledProcessError 异常。该异常的属性包含参数、退出代码以及标准输出和标准错误(如果它们被捕获)。

这个例子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 列表应按位置传递。因此,将剩余的参数作为关键字参数传递。

波本

什么时候Popen改用?我很难仅根据论点找到用例。但是,直接使用Popen将允许您访问其方法,包括poll'send_signal'、'terminate' 和 'wait'。

这是sourcePopen中给出的签名。我认为这是对信息的最精确封装(而不是):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将留给读者作为练习。

于 2017-10-18T16:37:52.680 回答
53

使用子进程

...或者对于一个非常简单的命令:

import os
os.system('cat testfile')
于 2008-09-18T01:43:30.940 回答
42

os.system还可以,但有点过时了。它也不是很安全。相反,尝试subprocess. subprocess不直接调用 sh,因此比os.system.

在此处获取更多信息。

于 2008-09-18T01:53:27.767 回答
39

还有铅球

>>> 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
于 2014-10-10T17:41:13.153 回答
38

2018 年 6 月 27 日发布的 Python 3.7.0 ( https://docs.python.org/3/whatsnew/3.7.html )开始,您可以以最强大但同样简单的方式实现您想要的结果。该答案旨在以简短的方式向您展示各种选项的基本摘要。有关深入的答案,请参阅其他答案。


TL;2021 年博士

的最大优点os.system(...)是它的简单性。subprocess更好并且仍然易于使用,尤其是从Python 3.5开始。

import subprocess
subprocess.run("ls -a", shell=True)

注意:这是您问题的确切答案 - 运行命令

像在贝壳里


首选方式

如果可能,删除 shell 开销并直接运行命令(需要列表)。

import subprocess
subprocess.run(["help"])
subprocess.run(["ls", "-a"])

在列表中传递程序参数。不要\"为包含空格的参数包含 -escaping。


高级用例

检查输出

以下代码不言自明:

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- 打开捕获。否则result.stderrresult.stdoutNone. 可从Python 3.7获得。

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)

标准错误应该在标准输出里面

您可能希望在标准输出中包含所有程序输出,甚至是错误。为此,运行

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. 使用 list ( ["ls", "-a"]) 或 set shell=True

类型错误:[...] NoneType [...]

检查您是否已设置capture_output=True.

TypeError:需要一个类似字节的对象,而不是 [...]

你总是从你的程序中收到字节结果。如果您想像普通字符串一样使用它,请设置text=True.

subprocess.CalledProcessError:命令“[...]”返回非零退出状态 1。

您的命令未成功运行。您可以禁用返回码检查或检查您的实际程序的有效性。

TypeError: init () got an unexpected keyword argument [...]

您可能使用的是早于 3.7.0 的 Python 版本;将其更新为可用的最新版本。否则,此 Stack Overflow 帖子中还有其他答案,向您展示了较旧的替代解决方案。

于 2020-10-13T19:20:51.310 回答
32

采用:

import os

cmd = 'ls -al'

os.system(cmd)

os - 该模块提供了一种使用操作系统相关功能的可移植方式。

对于更多os功能,这里是文档。

于 2015-06-29T11:34:22.260 回答
32

可以这么简单:

import os
cmd = "your command"
os.system(cmd)
于 2018-04-30T13:47:17.770 回答
26

这里还有一个之前没有提到的区别。

subprocess.Popen将 <command> 作为子进程执行。就我而言,我需要执行需要与另一个程序 <b> 通信的文件 <a>。

我尝试了子进程,执行成功。但是 <b> 无法与 <a> 通信。当我从终端运行两者时,一切正常。

还有一个:(注意:kwrite 的行为与其他应用程序不同。如果您使用 Firefox 尝试以下操作,结果将不一样。)

如果您尝试os.system("kwrite"),程序流程将冻结,直到用户关闭 kwrite。为了克服这一点,我尝试了os.system(konsole -e kwrite)。这次程序继续流动,但是kwrite变成了控制台的子进程。

任何运行 kwrite 的人都不是子进程(即在系统监视器中,它必须出现在树的最左边)。

于 2010-01-08T21:11:30.640 回答
26

os.system不允许您存储结果,因此如果您想将结果存储在某个列表或其他内容中,则subprocess.call可以使用。

于 2012-06-11T22:28:35.423 回答
25

subprocess.check_call如果您不想测试返回值,这很方便。它对任何错误都抛出异常。

于 2011-01-18T19:21:44.547 回答
25

我倾向于将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)
于 2014-04-30T14:37:04.533 回答
21

我为此编写了一个库shell.py

它现在基本上是 popen 和 shlex 的包装器。它还支持管道命令,因此您可以在 Python 中更轻松地链接命令。因此,您可以执行以下操作:

ex('echo hello shell.py') | "awk '{print $2}'"
于 2014-05-01T20:49:01.967 回答
19

在 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
于 2016-06-17T09:14:24.743 回答
19

在 Linux 下,如果您想调用将独立执行的外部命令(将在 Python 脚本终止后继续运行),您可以使用简单的队列作为任务假脱机程序at命令。

带有任务后台处理程序的示例:

import os
os.system('ts <your-command>')

关于任务后台处理程序 ( ts) 的注意事项:

  1. 您可以使用以下方式设置要运行的并发进程数(“插槽”):

    ts -S <number-of-slots>

  2. 安装ts不需要管理员权限。您可以使用简单的从源代码下载并编译它make,将其添加到您的路径中,您就完成了。

于 2016-11-27T00:15:34.280 回答
19

Invoke是一个 Python(2.7 和 3.4+)任务执行工具和库。它为运行 shell 命令提供了一个干净的高级 API:

>>> 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
于 2018-09-14T22:20:09.267 回答
18

您可以使用 Popen,然后您可以检查程序的状态:

from subprocess import Popen

proc = Popen(['ls', '-l'])
if proc.poll() is None:
    proc.kill()

查看subprocess.Popen

于 2012-07-16T15:16:24.527 回答
17

一个简单的方法是使用os 模块

import os
os.system('ls')

或者,您也可以使用 subprocess 模块:

import subprocess
subprocess.check_call('ls')

如果您希望将结果存储在变量中,请尝试:

import subprocess
r = subprocess.check_output('ls')
于 2014-08-24T21:46:12.373 回答
17

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 网表的输出

+--------------------------------------+------------+------+
| 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 |
+--------------------------------------+------------+------+

打印输出(networkId)

27a74fcd-37c0-4789-9414-9531b7e3f126
于 2016-07-20T09:50:01.300 回答
16

运行任何命令并返回结果的最简单方法:

from commands import getstatusoutput

try:
    return getstatusoutput("ls -ltr")
except Exception, e:
    return None
于 2012-07-25T06:51:50.050 回答
14

大多数情况:

在大多数情况下,您只需要这样的一小段代码

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)

于 2019-11-28T13:40:49.433 回答
13

通常,我将以下函数用于外部命令,这对于长时间运行的进程特别方便。以下方法在运行时跟踪进程输出 并返回输出,如果进程失败则引发异常。

如果该过程是在 process 上使用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])
于 2018-03-23T02:30:43.750 回答
12

这是我的两分钱:在我看来,这是处理外部命令时的最佳实践......

这些是执行方法的返回值...

pass, stdout, stderr = execute(["ls","-la"],"/home/user/desktop")

这是执行方法...

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]
于 2015-10-14T07:12:51.030 回答
11

只是为了增加讨论,如果您包括使用 Python 控制台,则可以从IPython调用外部命令。在 IPython 提示符下,您可以通过添加前缀 '!' 来调用 shell 命令。您还可以将 Python 代码与 shell 结合起来,并将 shell 脚本的输出分配给 Python 变量。

例如:

In [9]: mylist = !ls

In [10]: mylist
Out[10]:
['file1',
 'file2',
 'file3',]
于 2013-06-19T23:18:34.813 回答
11

我写了一个小库来帮助这个用例:

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找到

于 2018-10-30T11:40:43.637 回答
10

在 Python 中调用外部命令

调用外部命令的一种简单方法是使用os.system(...). 这个函数返回命令的退出值。但缺点是我们不会得到标准输出和标准错误。

ret = os.system('some_cmd.sh')
if ret != 0 :
    print 'some_cmd.sh execution returned failure'

在后台调用 Python 中的外部命令

subprocess.Popen为运行外部命令而不是使用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]
于 2018-04-04T06:57:44.277 回答
9

在 Python 中运行外部命令有很多不同的方法,它们都有自己的优点和缺点。

我和我的同事一直在编写 Python 系统管理工具,所以我们需要运行很多外部命令,有时你希望它们阻塞或异步运行、超时、每秒更新等。

也有不同的方式来处理返回码和错误,你可能想要解析输出,并提供新的输入(以一种期望的风格)。或者您将需要重定向标准输入标准输出标准错误以在不同的 tty 中运行(例如,当使用GNU Screen时)。

因此,您可能必须围绕外部命令编写很多包装器。所以这是我们编写的一个 Python 模块,它几乎可以处理您想要的任何东西,如果没有,它非常灵活,因此您可以轻松扩展它:

https://github.com/hpcugent/vsc-base/blob/master/lib/vsc/utils/run.py

它不能独立工作,需要我们的一些其他工具,并且多年来获得了许多专门的功能,因此它可能不是您的直接替代品,但它可以为您提供很多关于用于运行命令的 Python 内部如何工作以及如何处理某些情况的想法。

于 2013-04-17T14:10:06.270 回答
9

使用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(" "))
于 2014-04-12T11:58:23.700 回答
9

采用:

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',
 '']
于 2016-06-24T11:29:00.270 回答
9

例如(在 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。我提到这一点是因为在 tutorial.pdf(第 81 页)中有一个很好的 os 模块概述。此外,它是 Python 编码人员的权威资源。

于 2018-01-31T17:42:38.623 回答
9

对于subprocess在 Python 3.5+ 中使用,以下在 Linux 上为我做了诀窍:

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'))

文档中所述PIPE值是字节序列,为了正确显示它们,应该考虑解码。对于更高版本的 Python,text=True并被encoding='utf-8'添加到subprocess.run().

上述代码的输出是:

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
于 2019-01-29T05:11:40.887 回答
8

经过一些研究,我有以下代码对我来说效果很好。它基本上实时打印标准输出和标准错误。

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
于 2014-03-14T02:59:05.007 回答
8

这是调用外部命令并返回或打印命令的输出:

Python子进程check_output适用于

使用参数运行命令并将其输出作为字节字符串返回。

import subprocess
proc = subprocess.check_output('ipconfig /all')
print proc
于 2016-10-11T02:26:49.373 回答
8

如果您需要从 Python 笔记本(如Jupyter、Zeppelin、Databricks 或 Google Cloud Datalab)调用 shell 命令,您只需使用!前缀即可。

例如,

!ls -ilF
于 2018-05-08T20:49:55.577 回答
8

如果您正在编写 Python shell 脚本并在系统上安装了IPython,则可以使用 bang 前缀在 IPython 中运行 shell 命令:

!ls
filelist = !ls
于 2019-08-28T16:56:20.483 回答
6

对于 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)
于 2016-03-17T10:48:32.030 回答
5

2015 年更新:Python 3.5 添加了subprocess.run,它比 subprocess.Popen 更易于使用。我建议这样做。

>>> 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'')
于 2013-04-18T17:39:50.123 回答
5
import subprocess

p = subprocess.run(["ls", "-ltr"], capture_output=True)
print(p.stdout.decode(), p.stderr.decode())

在线尝试

于 2019-10-09T05:35:21.340 回答
4

使用 Python 模块的Popen功能subprocess是运行 Linux 命令的最简单方法。在那里,该Popen.communicate()函数将为您提供命令输出。例如

import subprocess

..
process = subprocess.Popen(..)   # Pass command and arguments to the function
stdout, stderr = process.communicate()   # Get command output and error
..
于 2015-07-24T19:12:21.037 回答
4

调用命令的方法有很多种。

  • 例如:

如果and.exe需要两个参数。在 cmd 我们可以调用sample.exeuse this: and.exe 2 3它会显示5在屏幕上。

如果我们使用 Python 脚本调用and.exe,我们应该这样做......

  1. os.system(cmd,...)

    • os.system(("and.exe" + " " + "2" + " " + "3"))
  2. os.popen(cmd,...)

    • os.popen(("and.exe" + " " + "2" + " " + "3"))
  3. subprocess.Popen(cmd,...)
    • subprocess.Popen(("and.exe" + " " + "2" + " " + "3"))

太难了,所以我们可以用空格加入cmd:

import os
cmd = " ".join(exename,parameters)
os.popen(cmd)
于 2016-09-12T09:44:09.617 回答
4

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
于 2019-11-26T11:39:48.007 回答
3

我会推荐以下方法“运行”,它将帮助我们获取标准输出标准错误和退出状态作为字典;它的调用者可以通过'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
于 2016-04-28T11:18:22.523 回答
3

如果你没有在命令中使用用户输入,你可以使用这个:

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将生成一个外壳,因此您可以使用管道和此类外壳的东西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 命令。

即使使用shlex.quote(). 在 shell 命令上使用用户输入时,最好保持一点偏执。一种选择是使用硬编码命令来获取一些通用输出并通过用户输入进行过滤。无论如何,使用shell=False将确保仅执行您要执行的确切进程,否则您会收到No such file or directory错误。

也有一些性能影响shell=True,从我的测试来看,它似乎比shell=False(默认)慢 20%。

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
于 2019-03-31T12:21:50.947 回答
3

Python 3.5+

import subprocess

p = subprocess.run(["ls", "-ltr"], capture_output=True)
print(p.stdout.decode(), p.stderr.decode())

在线尝试

于 2019-10-03T04:19:48.177 回答
3

有多种方法可以从 Python调用外部命令。有一些功能和模块具有很好的辅助功能,可以使它变得非常容易。但其中最推荐的是subprocess模块。

import subprocess as s
s.call(["command.exe", "..."])

调用函数将启动外部进程,传递一些命令行参数并等待它完成。完成后,您继续执行。call函数中的参数通过列表传递。列表中的第一个参数通常是可执行文件形式的命令,列表中的后续参数是您想要传递的任何内容。

如果您之前在 windows 的命令行中调用过进程,您会意识到您经常需要引用参数。你需要在它周围加上引号。如果有空格,则有反斜杠,并且有一些复杂的规则,但是您可以通过使用模块在 Python 中避免很多这样的情况,subprocess因为它是一个列表,并且每个项目都被认为是不同的,并且 python 可以正确引用你。

最后,在列表之后,有许多可选参数,其中之一是 shell,如果您将 shell 设置为 true,那么您的命令将像在命令提示符下键入一样运行。

s.call(["command.exe", "..."], shell=True)

这使您可以访问管道等功能,可以重定向到文件,可以在一件事中调用多个命令。

还有一件事,如果您的脚本依赖于成功的过程,那么您想要检查结果,并且可以使用 check call helper 函数检查结果。

s.check_call(...)

它与调用函数完全相同,它采用相同的参数,采用相同的列表,您可以传入任何额外的参数,但它会等待函数完成。如果函数的退出代码不是零,它将通过 python 脚本中的异常。

最后,如果您想要更严格的控制Popen构造函数,它也来自subprocess模块。它也采用与 incall 和 check_call 函数相同的参数,但它返回一个表示正在运行的进程的对象。

p=s.Popen("...")

它不会等待正在运行的进程完成,也不会立即抛出任何异常,但它会给你一个对象,让你做一些事情,比如等待它完成,让你与之通信,你可以重定向标准输入,标准输出,如果你想在其他地方显示输出等等。

于 2020-02-27T05:57:29.637 回答
2

我编写了一个包装器来处理错误并重定向输出和其他内容。

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)
于 2017-10-24T23:30:20.633 回答
1

Sultan是一个用于此目的的最新包。它提供了一些关于管理用户权限和添加有用的错误消息的细节。

from sultan.api import Sultan

with Sultan.load(sudo=True, hostname="myserver.com") as sultan:
  sultan.yum("install -y tree").run()
于 2019-07-01T20:46:19.407 回答
1

您可以使用子进程模块中的Popen运行任何命令。

from subprocess import Popen

首先,使用您要运行的所有参数创建命令对象。例如,在下面的代码片段中,gunicom 命令对象已由所有参数组成:

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()
于 2021-09-01T10:34:36.640 回答
0

2022年的答案。

这里有很多答案,但没有一个能满足我的所有需求。

  • 我需要运行命令并捕获输出退出代码
  • 我需要使执行的程序超时并在达到超时时强制它退出,并杀死它的所有子进程
  • 我需要它在winXP+cygwinlinux中工作。在 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 childrens when the timeout is reached (if parent did not detach)
    ## usage: _run(arg1, arg2, arg3)
        # arg1: command+args. 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 in windows xp, 7, cygwin and linux. 
    # - preexec_fn=os.setsid (py2) is equivalent to start_new_session (py3) (works in linux only), in windows and cygwin we use TASKKILL
    # - we use stderr=subprocess.STDOUT to merge stderr and stdout
    import sys, subprocess, os, signal, shlex, time
    
    def _runPY3(command, timeout_s=None, shell=False):
        # py3.3+ coz : 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:  # py2 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 但在大多数情况下使用此功能是无用的,我更喜欢选择我自己想要的 shell,但如果你也需要它,这里是:

# 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 中受支持,但在winxp3.7+中最后支持的 python 版本是.3.4
  • 并且因为这个函数的超时参数在windows中是没有用的,所以它不会杀死执行命令的子进程。
  • 如果使用capture_output+ timeoutarg,如果有子进程仍在运行,它将挂起。它仍然在 Windows 中被破坏,问题31447仍然存在。
于 2022-01-20T15:58:14.867 回答
-2

我将它用于 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
于 2020-04-19T16:01:21.957 回答