Как я могу асинхронно запустить внешнюю команду из Python?

Мне нужно запустить команду оболочки асинхронно из сценария Python. Под этим я подразумеваю, что я хочу, чтобы мой скрипт Python продолжал работать, пока внешняя команда отключается и делает все, что ей нужно.

Я прочитал этот пост:

Вызов внешней команды в Python

Затем я ушел и провел некоторое тестирование, и, похоже, os.system()он выполнит работу при условии, что я использую &в конце команды, чтобы мне не пришлось ждать ее возврата. Меня интересует, правильно ли это сделать? Я пробовал, commands.call()но у меня это не сработает, потому что блокируется внешняя команда.

Пожалуйста, дайте мне знать, os.system()рекомендуется ли использовать для этого или я должен попробовать другой путь.

subprocess.Popen делает именно то, что вы хотите.

from subprocess import Popen
p = Popen(['watch', 'ls']) # something long running
# ... do other stuff while subprocess is running
p.terminate()

(Отредактируйте, чтобы завершить ответ из комментариев)

Экземпляр Popen может делать разные другие вещи, например, вы можете poll()видеть, работает ли он, и вы можете communicate()с его помощью отправлять ему данные на стандартный ввод и ждать его завершения.

Если вы хотите запускать несколько процессов параллельно, а затем обрабатывать их, когда они дают результаты, вы можете использовать опрос, как показано ниже:

from subprocess import Popen, PIPE
import time

running_procs = [
    Popen(['/usr/bin/my_cmd', '-i %s' % path], stdout=PIPE, stderr=PIPE)
    for path in '/tmp/file0 /tmp/file1 /tmp/file2'.split()]

while running_procs:
    for proc in running_procs:
        retcode = proc.poll()
        if retcode is not None: # Process finished.
            running_procs.remove(proc)
            break
        else: # No process is done, wait a bit and check again.
            time.sleep(.1)
            continue

    # Here, `proc` has finished with return code `retcode`
    if retcode != 0:
        """Error handling."""
    handle_results(proc.stdout)

Поток управления там немного запутан, потому что я стараюсь сделать его небольшим - вы можете рефакторинг по своему вкусу. :-)

Это дает преимущество в том, что в первую очередь обслуживаются запросы раннего завершения. Если вы вызываете communicateпервый запущенный процесс, и он работает дольше всех, другие запущенные процессы будут бездействовать, тогда как вы могли бы обрабатывать их результаты.

Мне интересно, является ли эта [os.system ()] правильным способом для выполнения такой задачи?

Нет, os.system()это не правильный путь. Вот почему все говорят использовать subprocess.

Для получения дополнительной информации прочтите http://docs.python.org/library/os.html#os.system.

Модуль подпроцесса предоставляет более мощные средства для создания новых процессов и получения их результатов; использование этого модуля предпочтительнее, чем использование этой функции. Используйте модуль подпроцесса. Обратите особое внимание на раздел «Замена старых функций с помощью модуля подпроцесса».

У меня был хороший успех с модулем asyncproc , который хорошо справляется с выводом процессов. Например:

import os
from asynproc import Process
myProc = Process("myprogram.app")

while True:
    # check to see if process has ended
    poll = myProc.wait(os.WNOHANG)
    if poll is not None:
        break
    # print any new output
    out = myProc.read()
    if out != "":
        print out

Другой способ сделать это - использовать pexpect с неблокирующими строками чтения . Pexpect решает проблемы взаимоблокировки, позволяет легко запускать процессы в фоновом режиме и дает простые способы иметь обратные вызовы, когда ваш процесс выплевывает предопределенные строки, и в целом значительно упрощает взаимодействие с процессом.

Учитывая «мне не нужно ждать, пока он вернется», одним из самых простых решений будет следующее:

subprocess.Popen( \
    [path_to_executable, arg1, arg2, ... argN],
    creationflags = subprocess.CREATE_NEW_CONSOLE,
).pid

Но ... Из того, что я прочитал, это не «правильный способ сделать это» из-за рисков безопасности, создаваемых subprocess.CREATE_NEW_CONSOLEфлагом.

Ключевые вещи, которые здесь происходят, - это использование subprocess.CREATE_NEW_CONSOLEдля создания новой консоли и .pid(возвращает идентификатор процесса, чтобы вы могли позже проверить программу, если хотите), чтобы не ждать, пока программа завершит свою работу.

У меня такая же проблема при попытке подключиться к терминалу 3270 с помощью программного обеспечения сценариев s3270 на Python. Теперь я решаю проблему с подклассом Process, который нашел здесь:

http://code.activestate.com/recipes/440554/

А вот образец из файла:

def recv_some(p, t=.1, e=1, tr=5, stderr=0):
    if tr < 1:
        tr = 1
    x = time.time()+t
    y = []
    r = ''
    pr = p.recv
    if stderr:
        pr = p.recv_err
    while time.time() < x or r:
        r = pr()
        if r is None:
            if e:
                raise Exception(message)
            else:
                break
        elif r:
            y.append(r)
        else:
            time.sleep(max((x-time.time())/tr, 0))
    return ''.join(y)

def send_all(p, data):
    while len(data):
        sent = p.send(data)
        if sent is None:
            raise Exception(message)
        data = buffer(data, sent)

if __name__ == '__main__':
    if sys.platform == 'win32':
        shell, commands, tail = ('cmd', ('dir /w', 'echo HELLO WORLD'), '\r\n')
    else:
        shell, commands, tail = ('sh', ('ls', 'echo HELLO WORLD'), '\n')

    a = Popen(shell, stdin=PIPE, stdout=PIPE)
    print recv_some(a),
    for cmd in commands:
        send_all(a, cmd + tail)
        print recv_some(a),
    send_all(a, 'exit' + tail)
    print recv_some(a, e=0)
    a.wait()

Принятый ответ очень старый.

Я нашел здесь лучший современный ответ:

https://kevinmccarthy.org/2016/07/25/streaming-subprocess-stdin-and-stdout-with-asyncio-in-python/

и внес некоторые изменения:

1. заставить его работать на окнах
2. заставить его работать с несколькими командами

import sys
import asyncio

if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())


async def _read_stream(stream, cb):
    while True:
        line = await stream.readline()
        if line:
            cb(line)
        else:
            break


async def _stream_subprocess(cmd, stdout_cb, stderr_cb):
    try:
        process = await asyncio.create_subprocess_exec(
            *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
        )

        await asyncio.wait(
            [
                _read_stream(process.stdout, stdout_cb),
                _read_stream(process.stderr, stderr_cb),
            ]
        )
        rc = await process.wait()
        return process.pid, rc
    except OSError as e:
        # the program will hang if we let any exception propagate
        return e


def execute(*aws):
    """ run the given coroutines in an asyncio loop
    returns a list containing the values returned from each coroutine.
    """
    loop = asyncio.get_event_loop()
    rc = loop.run_until_complete(asyncio.gather(*aws))
    loop.close()
    return rc


def printer(label):
    def pr(*args, **kw):
        print(label, *args, **kw)

    return pr


def name_it(start=0, template="s{}"):
    """a simple generator for task names
    """
    while True:
        yield template.format(start)
        start += 1


def runners(cmds):
    """
    cmds is a list of commands to excecute as subprocesses
    each item is a list appropriate for use by subprocess.call
    """
    next_name = name_it().__next__
    for cmd in cmds:
        name = next_name()
        out = printer(f"{name}.stdout")
        err = printer(f"{name}.stderr")
        yield _stream_subprocess(cmd, out, err)


if __name__ == "__main__":
    cmds = (
        [
            "sh",
            "-c",
            """echo "$SHELL"-stdout && sleep 1 && echo stderr 1>&2 && sleep 1 && echo done""",
        ],
        [
            "bash",
            "-c",
            "echo 'hello, Dave.' && sleep 1 && echo dave_err 1>&2 && sleep 1 && echo done",
        ],
        [sys.executable, "-c", 'print("hello from python");import sys;sys.exit(2)'],
    )

    print(execute(*runners(cmds)))

Маловероятно, что приведенные в примере команды будут идеально работать в вашей системе и не будут обрабатывать странные ошибки, но этот код действительно демонстрирует один из способов запуска нескольких подпроцессов с использованием asyncio и потоковой передачи вывода.

Здесь есть несколько ответов, но ни один из них не удовлетворяет моим нижеприведенным требованиям:

1. Я не хочу ждать завершения команды или засорять мой терминал выводами подпроцесса.

2. Я хочу запустить сценарий bash с перенаправлением.

3. Я хочу поддерживать конвейер в моем сценарии bash (например find ... | tar ...).

Единственная комбинация, которая удовлетворяет вышеуказанным требованиям:

subprocess.Popen(['./my_script.sh "arg1" > "redirect/path/to"'],
                 stdout=subprocess.PIPE, 
                 stderr=subprocess.PIPE,
                 shell=True)

Это рассматривается в примерах подпроцесса Python 3 в разделе «Подождите, пока команда не завершится асинхронно»:

import asyncio

proc = await asyncio.create_subprocess_exec(
    'ls','-lha',
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.PIPE)

# do something else while ls is working

# if proc takes very long to complete, the CPUs are free to use cycles for 
# other processes
stdout, stderr = await proc.communicate()

Процесс начнется, как только await asyncio.create_subprocess_exec(...)будет завершен. Если к тому времени, когда вы позвоните await proc.communicate(), он не будет завершен , он будет ждать там, чтобы сообщить вам статус вашего вывода. Если он закончился, proc.communicate()немедленно вернусь.

Суть здесь аналогична ответу Террелса но я думаю, что ответ Террелса, похоже, слишком усложняет ситуацию.

См. asyncio.create_subprocess_execДополнительную информацию.