我常常用下列函数用于外部命令,这特别适用于长时间运行过程,下面的方法在运行时将过程输出缩短,并返回输出,如果过程失败,则会产生一个例外。

结果是,如果这个过程是通过 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])

经过一些研究,我有下面的代码,这对我来说非常好,它基本上在实时打印了标准输出和标准错误。

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

假设你想从一个 CGI 脚本开始一个漫长的任务,也就是说,孩子的过程应该比 CGI 脚本执行过程更长。

从子过程模块文档的经典例子是:

import subprocess
import sys

# Some code here

pid = subprocess.Popen([sys.executable, "longtask.py"]) # Call subprocess

# Some more code here

我的目标平台是FreeBSD,但开发是在Windows上,所以我首先面对了Windows上的问题。

在Windows(Windows XP)上,父母过程不会完成,直到 longtask.py 完成了工作。 这不是你想要的 CGI 脚本. 问题不是具体的 Python; 在 PHP 社区,问题是相同的。

解决方案是将 DETACHED_PROCESS Process Creation Flag 转移到 Windows API 中的 CreateProcess 功能。 如果您安装了 pywin32,您可以从 win32 过程模块中导入旗帜,否则您应该自己定义:

DETACHED_PROCESS = 0x00000008

pid = subprocess.Popen([sys.executable, "longtask.py"],
                       creationflags=DETACHED_PROCESS).pid

pid = subprocess.Popen([sys.executable, "longtask.py"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE)

我没有在其他平台上检查代码,也不知道FreeBSD的行为原因,如果有人知道的话,请分享你的想法。

查看Python图书馆的“快速”图书馆,也。

它允许对外程序 / 命令的互动控制,甚至 ssh, ftp, telnet 等。

child = pexpect.spawn('ftp 192.168.0.24')

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

child.sendline('anonymous')

child.expect('(?i)password')

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

通话函数将启动外部过程,通过一些命令线论点,等待它完成。当它完成时,您将继续执行.通话函数中的论点通过列表.列表中的第一个论点是命令通常以执行文件的形式,随后在列表中的论点,您想要通过什么。

最终,在列表后,有几个选项参数其中一个是阴影,如果你设置阴影等于真实,那么你的命令将运行,就像你在命令即时输入一样。

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

s.check_call(...)

最后,如果你想要更紧的控制Popen构建器,这也是从子过程模块。 它也需要相同的论点作为incall & check_call 函数,但它返回一个代表运行过程的对象。

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

如果您需要从 Python 笔记本电脑(如 Jupyter、Zeppelin、Databricks 或 Google Cloud Datalab)中拨打一个支票命令,您只能使用! 预定。

例如,

!ls -ilF