Live stdout output from Python subprocess in Jupyter notebook

The ipython notebook has it's own support for running shell commands. If you don't need to capture with subprocess stuff you can just do

cmd = 'ls -l'
!{cmd}

Output from commands executed with ! is automatically piped through the notebook.

If you set stdout = None (this is the default, so you can omit the stdout argument altogether), then your process should write its output to the terminal running your IPython notebook server.

This happens because the default behavior is for subprocess to inherit from the parent file handlers (see docs).

Your code would look like this:

from subprocess import Popen, PIPE
import shlex

def run_command(cmd):
    p = Popen(shlex.split(cmd), bufsize=1, universal_newlines=True)
    return p.poll()

This won't print to the notebook in browser, but at least you will be able to see the output from your subprocess asynchronously while other code is running.

Hope this helps.

Jupyter mucks with stdout and stderr. This should get what you want, and give you a more useful exception when the command fails to boot.

import signal
import subprocess as sp


class VerboseCalledProcessError(sp.CalledProcessError):
    def __str__(self):
        if self.returncode and self.returncode < 0:
            try:
                msg = "Command '%s' died with %r." % (
                    self.cmd, signal.Signals(-self.returncode))
            except ValueError:
                msg = "Command '%s' died with unknown signal %d." % (
                    self.cmd, -self.returncode)
        else:
            msg = "Command '%s' returned non-zero exit status %d." % (
                self.cmd, self.returncode)

        return f'{msg}\n' \
               f'Stdout:\n' \
               f'{self.output}\n' \
               f'Stderr:\n' \
               f'{self.stderr}'


def bash(cmd, print_stdout=True, print_stderr=True):
    proc = sp.Popen(cmd, stderr=sp.PIPE, stdout=sp.PIPE, shell=True, universal_newlines=True,
                    executable='/bin/bash')

    all_stdout = []
    all_stderr = []
    while proc.poll() is None:
        for stdout_line in proc.stdout:
            if stdout_line != '':
                if print_stdout:
                    print(stdout_line, end='')
                all_stdout.append(stdout_line)
        for stderr_line in proc.stderr:
            if stderr_line != '':
                if print_stderr:
                    print(stderr_line, end='', file=sys.stderr)
                all_stderr.append(stderr_line)

    stdout_text = ''.join(all_stdout)
    stderr_text = ''.join(all_stderr)
    if proc.wait() != 0:
        raise VerboseCalledProcessError(proc.returncode, cmd, stdout_text, stderr_text)

Replacing the for loop with the explicit readline() call worked for me.

from subprocess import Popen, PIPE
import shlex

def run_command(cmd):
    with Popen(shlex.split(cmd), stdout=PIPE, bufsize=1, universal_newlines=True) as p:
        while True:
            line = p.stdout.readline()
            if not line:
                break
            print(line)    
        exit_code = p.poll()
    return exit_code

Something is still broken about their iterators, even 4 years later.

Use the subprocess.check_output function:

>>> subprocess.check_output(['echo', 'foobar'])
b'foobar\n'

For Python 3 you get back a bytes object which you can decode:

>>> b=subprocess.check_output(['echo', 'foobar'])
>>> b.decode().strip()

'foobar'

If you want to treat stdout and stderr separately, you can spawn two threads that handle them concurrently (live as the output is produced). This works in Jupyter notebooks as well as plain python interpreters / scripts.

Adapted from my more detailed answer:

import logging
from collections import deque
from concurrent.futures import ThreadPoolExecutor
from functools import partial
from subprocess import PIPE, CalledProcessError, CompletedProcess, Popen


def stream_command(
    args,
    *,
    stdout_handler=logging.info,
    stderr_handler=logging.error,  # for a version that redirects stderr to stdout (without multithreading), see https://stackoverflow.com/a/76626021/5511061
    check=True,
    text=True,
    **kwargs,
):
    """Mimic subprocess.run, while processing the command output in real time ref https://stackoverflow.com/a/76634163/5511061."""
    with (
        Popen(args, text=text, stdout=PIPE, stderr=PIPE, **kwargs) as process,
        ThreadPoolExecutor(2) as pool,  # two threads to handle the (live) streams separately
    ):
        exhaust = partial(deque, maxlen=0)  # collections recipe: exhaust an iterable at C-speed
        exhaust_async = partial(pool.submit, exhaust)  # exhaust non-blocking in a background thread
        # exhaust both streams (stdout and stderr iterables) asynchronously, each in its own thread
        exhaust_async(stdout_handler(line[:-1]) for line in process.stdout)
        exhaust_async(stderr_handler(line[:-1]) for line in process.stderr)
    retcode = process.poll()  # block until both iterables are exhausted (process finished)
    if check and retcode:
        raise CalledProcessError(retcode, process.args)
    return CompletedProcess(process.args, retcode)

Call with simple print handlers:

stream_command(["echo", "test"], stdout_handler=print, stderr_handler=print)
# test

Or with custom handlers:

outs, errs = [], []
def stdout_handler(line):
    outs.append(line)
    print(line)
def stderr_handler(line):
    errs.append(line)
    print(line)

stream_command(
    ["echo", "test"],
    stdout_handler=stdout_handler,
    stderr_handler=stderr_handler,
)
# test
print(outs)
# ['test']