Python: Stream output of `asyncio.create_subprocess_exec()`

asyncio_streaming_example.py

"""Stream output of `asyncio.create_subprocess_exec()`"""
import asyncio
import sys


async def run(program: str, args: list[str]):
    """Capture output (stdout and stderr) while running external command."""
    proc = await asyncio.create_subprocess_exec(
        program, *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )

    while True:
        if proc.stdout.at_eof() and proc.stderr.at_eof():
            break

        stdout = (await proc.stdout.readline()).decode()
        if stdout:
            print(f'[stdout] {stdout}', end='', flush=True)
        stderr = (await proc.stderr.readline()).decode()
        if stderr:
            print(f'[sdterr] {stderr}', end='', flush=True, file=sys.stderr)

        await asyncio.sleep(1)

    await proc.communicate()

    print(f'{program} {" ".join(args)} exited with {proc.returncode}')


async def run2(program: str, args: list[str]):
    """Run external command without capturing output (stdout and stderr)."""
    proc = await asyncio.create_subprocess_exec(program, *args)
    await proc.communicate()


class Runner:
    """Run external command with output capturing.

    Usage:
        runner = Runner(program, args, interval)
        runner.start()

        async for stdout, stderr in runner.stream():
            ...

        returncode = runner.wait()
    """

    def __init__(self, program, args, interval=1):
        self.proc = None
        self.program = program
        self.args = args
        self.interval = interval

    async def start(self):
        self.proc = await asyncio.create_subprocess_exec(
            self.program,
            *self.args,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

    async def stream(self):
        while True:
            if self.proc.stdout.at_eof() and self.proc.stderr.at_eof():
                break

            stdout = await self.proc.stdout.readline()
            stderr = await self.proc.stderr.readline()
            yield stdout.decode(), stderr.decode()

            await asyncio.sleep(self.interval)

    async def wait(self):
        if self.proc is None:
            return None

        await self.proc.communicate()
        return self.proc.returncode


async def main():
    runner = Runner('sh', ['./never-ending-script.sh'])

    await runner.start()

    async for stdout, stderr in runner.stream():
        if stdout:
            print(f'[stdout] {stdout}', end='', flush=True)
        if stderr:
            print(f'[sdterr] {stderr}', end='', flush=True, file=sys.stderr)

    returncode = await runner.wait()
    print(f'return code is {returncode}')


if __name__ == '__main__':
    asyncio.run(run('sh', ['./never-ending-script.sh']))
    asyncio.run(run2('sh', ['./never-ending-script.sh']))
    asyncio.run(main())

should read stdout and stderr at the same time