# Execute command using asyncio.create_subprocess_shell and yield lines of stdout, stderr and finally the return code

My task is to run a subprocess using asyncio.create_subprocess_shell and yield the lines of output. I am using asyncio to avoid creating threads just to pump the streams.

• It is necessary to separate stdout lines from stderr lines.
• Both outputs need to be read simultaneously, it is not possible to read all of stdout or stderr first, because then the other output buffer in the subprocess may fill and it would stop running.
• This function is incorporated into a gRPC service, which is the reason for wrapping the yielded lines in instances of cli_pb2.ExecuteReply(...)
• .readline() returns "" on EOF, so when I see it, that is when to stop reading that stream.

I want to ask if my approach to reading stdout and stderr simultaneously can be improved. I was looking for something akin to UNIX select() and found asyncio.wait({...}, return_when=asyncio.FIRST_COMPLETED). Then I had to figure out which stream (stdout, stderr) the result corresponds to. I hacked at it by looking at _coro.

import asyncio

import cli_pb2
import cli_pb2_grpc

async def run(cmd):
proc = await asyncio.create_subprocess_shell(
cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)

while pending:
done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
for f in done:
line = (await f).decode()
if f._coro is stderr_readline:  # bad, but it works and can't think of better approach
if line:
pending.add(stderr_readline)  # put it back into the pending set
if line:
print(line)
await proc.wait()
status = proc.returncode

Usage example

async def printlines():
async for line in run("echo a; sleep 5; echo b; echo c"):
print(line)

outputs (with a 5 second pause after first line

stdout: "a\n"
stdout: "b\n"
stdout: "c\n"

The protobuf definition is

syntax = "proto3";

option java_multiple_files = true;
option java_package = "djtests.cli";
option java_outer_classname = "CliProto";

package cli;

service Executor {
rpc Execute (ExecuteRequest) returns (stream ExecuteReply) {
};
}

message ExecuteRequest {
string cmd = 1;
}

$$`$$