Subprocess Selector

10 Feb, 2023

Sometimes you want to be able to read stdout and stderr lines from a subprocess at the same time.

You can do it in Python like this:

import os
import subprocess
import shlex
import selectors


class ExecError(OSError):
    def __init__(self, exit_code, stdout, stderr, *k, **p):
        super().__init__(*k, **p)
        self.exit_code = exit_code
        self.stdout = stdout
        self.stderr = stderr


def exec(cmd, cwd=None, env=None):
    process = subprocess.Popen(
        cmd,
        cwd=cwd,
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        universal_newlines=True,
    )
    stdout_lines = []
    stderr_lines = []
    # Read both stdout and stderr simultaneously
    sel = selectors.DefaultSelector()
    assert process.stdout is not None
    assert process.stderr is not None
    sel.register(process.stdout, selectors.EVENT_READ)
    sel.register(process.stderr, selectors.EVENT_READ)
    stdout_ok = True
    stderr_ok = True
    while stdout_ok or stderr_ok:
        for key, _ in sel.select():
            line = key.fileobj.readline()  #  type: ignore
            if not line:
                if key.fileobj is process.stdout:
                    stdout_ok = False
                else:
                    stderr_ok = False
            elif stdout_ok or stderr_ok:
                if key.fileobj is process.stdout:
                    stdout_lines.append(line)
                else:
                    stderr_lines.append(line)
    # Calling this closes open files
    o, e = process.communicate()
    assert o == "", o
    assert e == "", e
    exit_code = process.wait()
    stdout = "\n".join(stdout_lines)
    stderr = "\n".join(stderr_lines)
    if exit_code != 0:
        raise ExecError(
            exit_code, stdout, stderr, f"Exec failed: {stderr or stdout}"
        )
    return stdout, stderr

if __name__ == '__main__':
    print(exec(['echo', 'hi']))

You can run it like this (although this doesn't demonstrate both stderr, you can see it works):

% python3 sp.py
('hi\n', '')

Comments

Be the first to comment.

Add Comment

Copyright James Gardner 1996-2020 All Rights Reserved. Admin.