python - asyncio を使用して子プロセスから stdout/stderr をストリーミングし、終了コードを取得する方法は?

Question

Windows 上の Python 3.4 では、子プロセスによって stdout/stderr に書き込まれたデータをストリーミングする必要があります。つまり、Python 3.4 で導入されたasyncioフレームワークを使用して、発生した出力を受け取ります。その後、プログラムの終了コードも特定する必要があります。これどうやってするの？

Answer 1

これまでに思いついたソリューションでは、SubprocessProtocolを使用して子プロセスから出力を受け取り、関連するトランスポートを使用してプロセスの終了コードを取得します。これが最適かどうかはわかりませんが。JF Sebastian による同様の質問への回答に基づいてアプローチしました。

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

Answer 2

イベント ループは、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()