Как передать поток stdout/stderr из дочернего процесса с помощью asyncio и получить его код выхода после?

Решение, которое я придумал до сих пор, использует SubprocessProtocol для получения вывод дочернего процесса и связанный с ним транспорт для получения кода завершения процесса. Хотя не знаю, оптимально ли это. Я основывал свой подход на ответе на аналогичный вопрос Дж. Ф. Себастьяна.

import asyncio
import contextlib
import os
import locale


class SubprocessProtocol(asyncio.SubprocessProtocol):
    def pipe_data_received(self, fd, data):
        if fd == 1:
            name = 'stdout'
        elif fd == 2:
            name = 'stderr'
        text = data.decode(locale.getpreferredencoding(False))
        print('Received from {}: {}'.format(name, text.strip()))

    def process_exited(self):
        loop.stop()


if os.name == 'nt':
    # On Windows, the ProactorEventLoop is necessary to listen on pipes
    loop = asyncio.ProactorEventLoop()
    asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()
with contextlib.closing(loop):
    # This will only connect to the process
    transport = loop.run_until_complete(loop.subprocess_exec(
        SubprocessProtocol, 'python', '-c', 'print(\'Hello async world!\')'))[0]
    # Wait until process has finished
    loop.run_forever()
    print('Program exited with: {}'.format(transport.get_returncode()))

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

Это исправление для ответа aknuds1:

class SubprocessProtocol(asyncio.SubprocessProtocol):
    def __init__(self):
        self._exited = False
        self._closed_stdout = False
        self._closed_stderr = False

    @property
    def finished(self):
        return self._exited and self._closed_stdout and self._closed_stderr

    def signal_exit(self):
        if not self.finished:
            return
        loop.stop()        

    def pipe_data_received(self, fd, data):
        if fd == 1:
            name = 'stdout'
        elif fd == 2:
            name = 'stderr'
        text = data.decode(locale.getpreferredencoding(False))
        print('Received from {}: {}'.format(name, text.strip()))

    def pipe_connection_lost(self, fd, exc):
        if fd == 1:
            self._closed_stdout = True
        elif fd == 2:
            self._closed_stderr = True
        self.signal_exit()

    def process_exited(self):
        self._exited = True
        self.signal_exit()

Я предполагаю использовать API высокого уровня:

proc = yield from asyncio.create_subprocess_exec(
    'python', '-c', 'print(\'Hello async world!\')')

stdout, stderr = yield from proc.communicate()

retcode = proc.returncode

Также вы можете сделать больше:

yield from proc.stdin.write(b'data')
yield from proc.stdin.drain()

stdout = yield from proc.stdout.read()
stderr = yield from proc.stderr.read()

retcode = yield from proc.wait()

и так далее.

Но, пожалуйста, имейте в виду, что ожидание, скажем, stdout, когда дочерний процесс ничего не напечатает, может повесить вашу сопрограмму.