Experiments in capturing combined output streams from subprocesses in Python.

stdout_stderr_combined.py

# Experiments in capturing combined output streams from subprocesses.
# 
# It turns out it's kind of hard to get the interleaved results of stdout and
# stderr in Python. However, in a lot of situations where Python is calling
# out to other processes, you probably want to swallow the child process's
# stderr when things to right and print it when things go wrong. Since some
# programs may be outputting results on stdout and warnings and errors on
# stderr, it makes sense that you'd want it all, and all in the order it was
# printed when things go wrong.
# 
# (Caveat all kinds of buffering can be happening in the system, and if stdout
# and stderr are being written to in rapid succession)
import os
import subprocess
import sys

# An implementation using selectors (fancy!)
import selectors
import io
def run_and_capture(command):
    child = subprocess.Popen(command,
                             stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE)

    combined = io.BytesIO()
    stdout = io.BytesIO()
    stderr = io.BytesIO()
    selector = selectors.DefaultSelector()
    selector.register(child.stdout, selectors.EVENT_READ)
    selector.register(child.stderr, selectors.EVENT_READ)

    while True:
        data = None
        for output, _ in selector.select():
            data = output.fileobj.read1()
            if data:
                combined.write(data)
                if output.fileobj is child.stdout:
                    stdout.write(data)
                else:
                    stderr.write(data)

        if not data:
            break
    
    # We exhausted stdout/stderr, but the child hasn't actually exited yet.
    child.wait()
    selector.close()
    return (child,
            stdout.getvalue().decode('utf-8'),
            stderr.getvalue().decode('utf-8'),
            combined.getvalue().decode('utf-8'))


# A more traditional implementation using threads.
import io
import threading
def run_and_capture2(command):
    combined_lock = threading.Lock()
    def stream_reader(read_stream, output_buffer, combined_buffer):
        while True:
            data = read_stream.read1()
            if not data:
                break
            output_buffer.write(data)
            with combined_lock:
                combined_buffer.write(data)

    stdout = io.BytesIO()
    stderr = io.BytesIO()
    combined = io.BytesIO()
    child = subprocess.Popen(command,
                             stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE)
    stdout_reader = threading.Thread(target=stream_reader, args=[child.stdout,
                                                                 stdout,
                                                                 combined])
    stderr_reader = threading.Thread(target=stream_reader, args=[child.stderr,
                                                                 stderr,
                                                                 combined])
    stdout_reader.start()
    stderr_reader.start()
    child.wait()
    stdout_reader.join()
    stderr_reader.join()
    
    return (child,
            stdout.getvalue().decode('utf-8'),
            stderr.getvalue().decode('utf-8'),
            combined.getvalue().decode('utf-8'))


child, result, _, all_output = run_and_capture(['./command-that-fails'])
print(f'returncode: {child.returncode}')
if child.returncode != 0:
    print(f'Error output: {all_output}')
    sys.exit(1)

print(f'Got result: "{result}"')

read1!!! You saved me