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

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

import os
os.system('ls')

或者，您还可以使用subprocess模块：

import subprocess
subprocess.check_call('ls')

如果您想将结果存储在变量中，请尝试：

import subprocess
r = subprocess.check_output('ls')

要从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

运行任何命令并获取结果的最简单方法：

from commands import getstatusoutput

try:
    return getstatusoutput("ls -ltr")
except Exception, e:
    return None

大多数情况:

对于大多数情况，您只需要像这样的一小段代码：

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

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)

我经常使用以下函数来运行外部命令，特别适用于长时间运行的进程。下面的方法在进程正在运行时跟踪进程输出并返回输出结果，如果进程失败，则引发异常。

当使用进程的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])

这是我的建议：在处理外部命令时，我认为这是最佳实践...

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

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]

我编写了一个小型库来帮助这种使用情况:

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控制台，可以从IPython调用外部命令。在IPython提示符中，您可以通过前缀'!'来调用shell命令。您还可以在Python代码中与shell结合，并将shell脚本的输出分配给Python变量。

例如：

In [9]: mylist = !ls

In [10]: mylist
Out[10]:
['file1',
 'file2',
 'file3',]

在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]

import subprocess
subprocess.run('mkdir test.dir', shell=True)

import subprocess
subprocess.call('mkdir test.dir', shell=True)

使用os.system的等效代码是：

import os
os.system('mkdir test.dir')